opal/mca/pmix/pmix4x/pmix/src/tool/pmix

/* [<][>][^][v][top][bottom][index][help] */
This source file includes following definitions.
_notify_complete
pmix_tool_notify_recv
tool_iof_handler
job_data
PMIx_tool_init
pmix_tool_init_info
fin_timeout
finwait_cbfunc
PMIx_tool_finalize
PMIx_tool_connect_to_server
   1 /* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
   2 /*
   3  * Copyright (c) 2014-2019 Intel, Inc.  All rights reserved.
   4  * Copyright (c) 2014-2019 Research Organization for Information Science
   5  *                         and Technology (RIST).  All rights reserved.
   6  * Copyright (c) 2014      Artem Y. Polyakov <artpol84@gmail.com>.
   7  *                         All rights reserved.
   8  * Copyright (c) 2016      Mellanox Technologies, Inc.
   9  *                         All rights reserved.
  10  * Copyright (c) 2016      IBM Corporation.  All rights reserved.
  11  * $COPYRIGHT$
  12  *
  13  * Additional copyrights may follow
  14  *
  15  * $HEADER$
  16  */
  17 
  18 #include <src/include/pmix_config.h>
  19 
  20 #include <src/include/pmix_socket_errno.h>
  21 
  22 #include "src/client/pmix_client_ops.h"
  23 #include <pmix_tool.h>
  24 #include <pmix_rename.h>
  25 
  26 #include "src/include/pmix_globals.h"
  27 
  28 #ifdef HAVE_STRING_H
  29 #include <string.h>
  30 #endif
  31 #include <fcntl.h>
  32 #ifdef HAVE_UNISTD_H
  33 #include <unistd.h>
  34 #endif
  35 #ifdef HAVE_SYS_SOCKET_H
  36 #include <sys/socket.h>
  37 #endif
  38 #ifdef HAVE_SYS_UN_H
  39 #include <sys/un.h>
  40 #endif
  41 #ifdef HAVE_SYS_UIO_H
  42 #include <sys/uio.h>
  43 #endif
  44 #ifdef HAVE_SYS_TYPES_H
  45 #include <sys/types.h>
  46 #endif
  47 #ifdef HAVE_DIRENT_H
  48 #include <dirent.h>
  49 #endif  /* HAVE_DIRENT_H */
  50 
  51 #include "src/class/pmix_list.h"
  52 #include "src/util/argv.h"
  53 #include "src/util/error.h"
  54 #include "src/util/hash.h"
  55 #include "src/util/output.h"
  56 #include "src/util/pmix_environ.h"
  57 #include "src/util/show_help.h"
  58 #include "src/runtime/pmix_progress_threads.h"
  59 #include "src/runtime/pmix_rte.h"
  60 #include "src/mca/bfrops/base/base.h"
  61 #include "src/mca/gds/base/base.h"
  62 #include "src/mca/pnet/base/base.h"
  63 #include "src/mca/ptl/base/base.h"
  64 #include "src/mca/psec/psec.h"
  65 #include "src/include/pmix_globals.h"
  66 #include "src/common/pmix_attributes.h"
  67 #include "src/common/pmix_iof.h"
  68 #include "src/server/pmix_server_ops.h"
  69 
  70 #define PMIX_MAX_RETRIES 10
  71 
  72 extern pmix_client_globals_t pmix_client_globals;
  73 static pmix_event_t stdinsig;
  74 static pmix_iof_read_event_t stdinev;
  75 
  76 static void _notify_complete(pmix_status_t status, void *cbdata)
  77 {
  78     pmix_event_chain_t *chain = (pmix_event_chain_t*)cbdata;
  79     PMIX_RELEASE(chain);
  80 }
  81 
  82 static void pmix_tool_notify_recv(struct pmix_peer_t *peer,
  83                                   pmix_ptl_hdr_t *hdr,
  84                                   pmix_buffer_t *buf, void *cbdata)
  85 {
  86     pmix_status_t rc;
  87     int32_t cnt;
  88     pmix_cmd_t cmd;
  89     pmix_event_chain_t *chain;
  90     size_t ninfo;
  91 
  92     pmix_output_verbose(2, pmix_client_globals.event_output,
  93                         "pmix:tool_notify_recv - processing event");
  94 
  95     /* a zero-byte buffer indicates that this recv is being
  96      * completed due to a lost connection */
  97     if (PMIX_BUFFER_IS_EMPTY(buf)) {
  98         return;
  99     }
 100 
 101       /* start the local notification chain */
 102     chain = PMIX_NEW(pmix_event_chain_t);
 103     chain->final_cbfunc = _notify_complete;
 104     chain->final_cbdata = chain;
 105 
 106     cnt=1;
 107     PMIX_BFROPS_UNPACK(rc, pmix_client_globals.myserver,
 108                        buf, &cmd, &cnt, PMIX_COMMAND);
 109     if (PMIX_SUCCESS != rc) {
 110         PMIX_ERROR_LOG(rc);
 111         PMIX_RELEASE(chain);
 112         goto error;
 113     }
 114     /* unpack the status */
 115     cnt=1;
 116     PMIX_BFROPS_UNPACK(rc, pmix_client_globals.myserver,
 117                        buf, &chain->status, &cnt, PMIX_STATUS);
 118     if (PMIX_SUCCESS != rc) {
 119         PMIX_ERROR_LOG(rc);
 120         PMIX_RELEASE(chain);
 121         goto error;
 122     }
 123 
 124     /* unpack the source of the event */
 125     cnt=1;
 126     PMIX_BFROPS_UNPACK(rc, pmix_client_globals.myserver,
 127                        buf, &chain->source, &cnt, PMIX_PROC);
 128     if (PMIX_SUCCESS != rc) {
 129         PMIX_ERROR_LOG(rc);
 130         PMIX_RELEASE(chain);
 131         goto error;
 132     }
 133 
 134     /* unpack the info that might have been provided */
 135     cnt=1;
 136     PMIX_BFROPS_UNPACK(rc, pmix_client_globals.myserver,
 137                        buf, &ninfo, &cnt, PMIX_SIZE);
 138     if (PMIX_SUCCESS != rc) {
 139         PMIX_ERROR_LOG(rc);
 140         PMIX_RELEASE(chain);
 141         goto error;
 142     }
 143 
 144     /* we always leave space for event hdlr name and a callback object */
 145     chain->nallocated = ninfo + 2;
 146     PMIX_INFO_CREATE(chain->info, chain->nallocated);
 147     if (NULL == chain->info) {
 148         PMIX_ERROR_LOG(PMIX_ERR_NOMEM);
 149         PMIX_RELEASE(chain);
 150         return;
 151     }
 152 
 153     if (0 < ninfo) {
 154         chain->ninfo = ninfo;
 155         cnt = ninfo;
 156         PMIX_BFROPS_UNPACK(rc, pmix_client_globals.myserver,
 157                            buf, chain->info, &cnt, PMIX_INFO);
 158         if (PMIX_SUCCESS != rc) {
 159             PMIX_ERROR_LOG(rc);
 160             PMIX_RELEASE(chain);
 161             goto error;
 162         }
 163     }
 164     /* prep the chain for processing */
 165     pmix_prep_event_chain(chain, chain->info, ninfo, false);
 166 
 167     pmix_output_verbose(2, pmix_client_globals.event_output,
 168                         "[%s:%d] pmix:tool_notify_recv - processing event %s from source %s:%d, calling errhandler",
 169                         pmix_globals.myid.nspace, pmix_globals.myid.rank, PMIx_Error_string(chain->status),
 170                         chain->source.nspace, chain->source.rank);
 171 
 172     pmix_invoke_local_event_hdlr(chain);
 173     return;
 174 
 175     error:
 176     /* we always need to return */
 177     pmix_output_verbose(2, pmix_client_globals.event_output,
 178                         "pmix:tool_notify_recv - unpack error status =%d, calling def errhandler", rc);
 179     chain = PMIX_NEW(pmix_event_chain_t);
 180     chain->status = rc;
 181     pmix_invoke_local_event_hdlr(chain);
 182 }
 183 
 184 static void tool_iof_handler(struct pmix_peer_t *pr,
 185                              pmix_ptl_hdr_t *hdr,
 186                              pmix_buffer_t *buf, void *cbdata)
 187 {
 188     pmix_peer_t *peer = (pmix_peer_t*)pr;
 189     pmix_proc_t source;
 190     pmix_iof_channel_t channel;
 191     pmix_byte_object_t bo;
 192     int32_t cnt;
 193     pmix_status_t rc;
 194 
 195     pmix_output_verbose(2, pmix_client_globals.iof_output,
 196                         "recvd IOF with %d bytes", (int)buf->bytes_used);
 197 
 198     /* if the buffer is empty, they are simply closing the channel */
 199     if (0 == buf->bytes_used) {
 200         return;
 201     }
 202 
 203     cnt = 1;
 204     PMIX_BFROPS_UNPACK(rc, peer, buf, &source, &cnt, PMIX_PROC);
 205     if (PMIX_SUCCESS != rc) {
 206         PMIX_ERROR_LOG(rc);
 207         return;
 208     }
 209     cnt = 1;
 210     PMIX_BFROPS_UNPACK(rc, peer, buf, &channel, &cnt, PMIX_IOF_CHANNEL);
 211     if (PMIX_SUCCESS != rc) {
 212         PMIX_ERROR_LOG(rc);
 213         return;
 214     }
 215     cnt = 1;
 216     PMIX_BFROPS_UNPACK(rc, peer, buf, &bo, &cnt, PMIX_BYTE_OBJECT);
 217     if (PMIX_SUCCESS != rc) {
 218         PMIX_ERROR_LOG(rc);
 219         return;
 220     }
 221     if (NULL != bo.bytes && 0 < bo.size) {
 222         pmix_iof_write_output(&source, channel, &bo, NULL);
 223     }
 224     PMIX_BYTE_OBJECT_DESTRUCT(&bo);
 225 }
 226 
 227 /* callback to receive job info */
 228 static void job_data(struct pmix_peer_t *pr,
 229                      pmix_ptl_hdr_t *hdr,
 230                      pmix_buffer_t *buf, void *cbdata)
 231 {
 232     pmix_status_t rc;
 233     char *nspace;
 234     int32_t cnt = 1;
 235     pmix_cb_t *cb = (pmix_cb_t*)cbdata;
 236 
 237     /* unpack the nspace - should be same as our own */
 238     PMIX_BFROPS_UNPACK(rc, pmix_client_globals.myserver,
 239                        buf, &nspace, &cnt, PMIX_STRING);
 240     if (PMIX_SUCCESS != rc) {
 241         PMIX_ERROR_LOG(rc);
 242         cb->status = PMIX_ERROR;
 243         PMIX_POST_OBJECT(cb);
 244         PMIX_WAKEUP_THREAD(&cb->lock);
 245         return;
 246     }
 247 
 248     /* decode it */
 249     PMIX_GDS_STORE_JOB_INFO(cb->status,
 250                             pmix_client_globals.myserver,
 251                             nspace, buf);
 252     cb->status = PMIX_SUCCESS;
 253     PMIX_POST_OBJECT(cb);
 254     PMIX_WAKEUP_THREAD(&cb->lock);
 255 }
 256 
 257 PMIX_EXPORT int PMIx_tool_init(pmix_proc_t *proc,
 258                                pmix_info_t info[], size_t ninfo)
 259 {
 260     pmix_status_t rc;
 261     char *evar, *nspace = NULL;
 262     pmix_rank_t rank = PMIX_RANK_UNDEF;
 263     bool gdsfound, do_not_connect = false;
 264     bool nspace_given = false;
 265     bool nspace_in_enviro = false;
 266     bool rank_given = false;
 267     bool fwd_stdin = false;
 268     pmix_info_t ginfo;
 269     size_t n;
 270     pmix_ptl_posted_recv_t *rcv;
 271     pmix_proc_t wildcard;
 272     int fd;
 273     pmix_proc_type_t ptype;
 274     pmix_cb_t cb;
 275     pmix_buffer_t *req;
 276     pmix_cmd_t cmd = PMIX_REQ_CMD;
 277 
 278     PMIX_ACQUIRE_THREAD(&pmix_global_lock);
 279 
 280     if (NULL == proc) {
 281         PMIX_RELEASE_THREAD(&pmix_global_lock);
 282         return PMIX_ERR_BAD_PARAM;
 283     }
 284     if (0 < pmix_globals.init_cntr) {
 285         /* since we have been called before, the nspace and
 286          * rank should be known. So return them here if
 287          * requested */
 288         if (NULL != proc) {
 289             pmix_strncpy(proc->nspace, pmix_globals.myid.nspace, PMIX_MAX_NSLEN);
 290             proc->rank = pmix_globals.myid.rank;
 291         }
 292         ++pmix_globals.init_cntr;
 293         PMIX_RELEASE_THREAD(&pmix_global_lock);
 294         return PMIX_SUCCESS;
 295     }
 296 
 297     /* parse the input directives */
 298     gdsfound = false;
 299     ptype = PMIX_PROC_TOOL;
 300     if (NULL != info) {
 301         for (n=0; n < ninfo; n++) {
 302             if (0 == strncmp(info[n].key, PMIX_GDS_MODULE, PMIX_MAX_KEYLEN)) {
 303                 PMIX_INFO_LOAD(&ginfo, PMIX_GDS_MODULE, info[n].value.data.string, PMIX_STRING);
 304                 gdsfound = true;
 305             } else if (0 == strncmp(info[n].key, PMIX_TOOL_DO_NOT_CONNECT, PMIX_MAX_KEYLEN)) {
 306                 do_not_connect = PMIX_INFO_TRUE(&info[n]);
 307             } else if (0 == strncmp(info[n].key, PMIX_TOOL_NSPACE, PMIX_MAX_KEYLEN)) {
 308                 if (NULL != nspace) {
 309                     /* cannot define it twice */
 310                     free(nspace);
 311                     if (gdsfound) {
 312                         PMIX_INFO_DESTRUCT(&ginfo);
 313                     }
 314                     PMIX_RELEASE_THREAD(&pmix_global_lock);
 315                     return PMIX_ERR_BAD_PARAM;
 316                 }
 317                 nspace = strdup(info[n].value.data.string);
 318                 nspace_given = true;
 319             } else if (0 == strncmp(info[n].key, PMIX_TOOL_RANK, PMIX_MAX_KEYLEN)) {
 320                 rank = info[n].value.data.rank;
 321                 rank_given = true;
 322             } else if (0 == strncmp(info[n].key, PMIX_FWD_STDIN, PMIX_MAX_KEYLEN)) {
 323                 /* they want us to forward our stdin to someone */
 324                 fwd_stdin = true;
 325             } else if (0 == strncmp(info[n].key, PMIX_LAUNCHER, PMIX_MAX_KEYLEN)) {
 326                 ptype |= PMIX_PROC_LAUNCHER;
 327             } else if (0 == strncmp(info[n].key, PMIX_SERVER_TMPDIR, PMIX_MAX_KEYLEN)) {
 328                 pmix_server_globals.tmpdir = strdup(info[n].value.data.string);
 329             } else if (0 == strncmp(info[n].key, PMIX_SYSTEM_TMPDIR, PMIX_MAX_KEYLEN)) {
 330                 pmix_server_globals.system_tmpdir = strdup(info[n].value.data.string);
 331             }
 332         }
 333     }
 334     if (NULL == pmix_server_globals.tmpdir) {
 335         if (NULL == (evar = getenv("PMIX_SERVER_TMPDIR"))) {
 336             pmix_server_globals.tmpdir = strdup(pmix_tmp_directory());
 337         } else {
 338             pmix_server_globals.tmpdir = strdup(evar);
 339         }
 340     }
 341     if (NULL == pmix_server_globals.system_tmpdir) {
 342         if (NULL == (evar = getenv("PMIX_SYSTEM_TMPDIR"))) {
 343             pmix_server_globals.system_tmpdir = strdup(pmix_tmp_directory());
 344         } else {
 345             pmix_server_globals.system_tmpdir = strdup(evar);
 346         }
 347     }
 348 
 349     if ((nspace_given && !rank_given) ||
 350         (!nspace_given && rank_given)) {
 351         /* can't have one and not the other */
 352         PMIX_ERROR_LOG(PMIX_ERR_BAD_PARAM);
 353         if (NULL != nspace) {
 354             free(nspace);
 355         }
 356         if (gdsfound) {
 357             PMIX_INFO_DESTRUCT(&ginfo);
 358         }
 359         PMIX_RELEASE_THREAD(&pmix_global_lock);
 360         return PMIX_ERR_BAD_PARAM;
 361     }
 362 
 363     /* if we were not passed an nspace in the info keys,
 364      * check to see if we were given one in the env - this
 365      * will be the case when we are launched by a PMIx-enabled
 366      * daemon */
 367     if (!nspace_given) {
 368         if (NULL != (evar = getenv("PMIX_NAMESPACE"))) {
 369             nspace = strdup(evar);
 370             nspace_in_enviro = true;
 371         }
 372     }
 373     /* also look for the rank - it normally is zero, but if we
 374      * were launched, then it might have been as part of a
 375      * multi-process tool */
 376     if (!rank_given) {
 377         if (NULL != (evar = getenv("PMIX_RANK"))) {
 378             rank = strtol(evar, NULL, 10);
 379             if (!nspace_in_enviro) {
 380                 /* this is an error - we can't have one and not
 381                  * the other */
 382                 PMIX_ERROR_LOG(PMIX_ERR_BAD_PARAM);
 383                 PMIX_RELEASE_THREAD(&pmix_global_lock);
 384                 return PMIX_ERR_BAD_PARAM;
 385             }
 386             /* flag that this tool is also a client */
 387             ptype |= PMIX_PROC_CLIENT_TOOL;
 388         } else if (nspace_in_enviro) {
 389             /* this is an error - we can't have one and not
 390              * the other */
 391             PMIX_ERROR_LOG(PMIX_ERR_BAD_PARAM);
 392             if (NULL != nspace) {
 393                 free(nspace);
 394             }
 395             if (gdsfound) {
 396                 PMIX_INFO_DESTRUCT(&ginfo);
 397             }
 398             PMIX_RELEASE_THREAD(&pmix_global_lock);
 399             return PMIX_ERR_BAD_PARAM;
 400         }
 401     }
 402 
 403     /* if we are a launcher, then we also need to act as a server,
 404      * so setup the server-related structures here */
 405     if (PMIX_PROC_LAUNCHER_ACT & ptype) {
 406         if (PMIX_SUCCESS != (rc = pmix_server_initialize())) {
 407             PMIX_ERROR_LOG(rc);
 408             if (NULL != nspace) {
 409                 free(nspace);
 410             }
 411             if (gdsfound) {
 412                 PMIX_INFO_DESTRUCT(&ginfo);
 413             }
 414             PMIX_RELEASE_THREAD(&pmix_global_lock);
 415             return rc;
 416         }
 417         /* setup the function pointers */
 418         memset(&pmix_host_server, 0, sizeof(pmix_server_module_t));
 419     }
 420 
 421     /* setup the runtime - this init's the globals,
 422      * opens and initializes the required frameworks */
 423     if (PMIX_SUCCESS != (rc = pmix_rte_init(ptype, info, ninfo,
 424                                             pmix_tool_notify_recv))) {
 425         PMIX_ERROR_LOG(rc);
 426         if (NULL != nspace) {
 427             free(nspace);
 428         }
 429         if (gdsfound) {
 430             PMIX_INFO_DESTRUCT(&ginfo);
 431         }
 432         PMIX_RELEASE_THREAD(&pmix_global_lock);
 433         return rc;
 434     }
 435     /* if we were given a name, then set it now */
 436     if (nspace_given || nspace_in_enviro) {
 437         pmix_strncpy(pmix_globals.myid.nspace, nspace, PMIX_MAX_NSLEN);
 438         free(nspace);
 439         pmix_globals.myid.rank = rank;
 440     }
 441 
 442     /* setup the IO Forwarding recv */
 443     rcv = PMIX_NEW(pmix_ptl_posted_recv_t);
 444     rcv->tag = PMIX_PTL_TAG_IOF;
 445     rcv->cbfunc = tool_iof_handler;
 446     /* add it to the end of the list of recvs */
 447     pmix_list_append(&pmix_ptl_globals.posted_recvs, &rcv->super);
 448 
 449 
 450     /* setup the globals */
 451     PMIX_CONSTRUCT(&pmix_client_globals.pending_requests, pmix_list_t);
 452     PMIX_CONSTRUCT(&pmix_client_globals.peers, pmix_pointer_array_t);
 453     pmix_pointer_array_init(&pmix_client_globals.peers, 1, INT_MAX, 1);
 454     pmix_client_globals.myserver = PMIX_NEW(pmix_peer_t);
 455     if (NULL == pmix_client_globals.myserver) {
 456         if (gdsfound) {
 457             PMIX_INFO_DESTRUCT(&ginfo);
 458         }
 459         PMIX_RELEASE_THREAD(&pmix_global_lock);
 460         return PMIX_ERR_NOMEM;
 461     }
 462     pmix_client_globals.myserver->nptr = PMIX_NEW(pmix_namespace_t);
 463     if (NULL == pmix_client_globals.myserver->nptr) {
 464         PMIX_RELEASE(pmix_client_globals.myserver);
 465         if (gdsfound) {
 466             PMIX_INFO_DESTRUCT(&ginfo);
 467         }
 468         PMIX_RELEASE_THREAD(&pmix_global_lock);
 469         return PMIX_ERR_NOMEM;
 470     }
 471     pmix_client_globals.myserver->info = PMIX_NEW(pmix_rank_info_t);
 472     if (NULL == pmix_client_globals.myserver->info) {
 473         PMIX_RELEASE(pmix_client_globals.myserver);
 474         if (gdsfound) {
 475             PMIX_INFO_DESTRUCT(&ginfo);
 476         }
 477         PMIX_RELEASE_THREAD(&pmix_global_lock);
 478         return PMIX_ERR_NOMEM;
 479     }
 480 
 481     pmix_output_verbose(2, pmix_globals.debug_output,
 482                         "pmix: init called");
 483 
 484     if (PMIX_PROC_IS_CLIENT(pmix_globals.mypeer)) {
 485         /* if we are a client, then we need to pickup the
 486          * rest of the envar-based server assignments */
 487         pmix_globals.pindex = -1;
 488         /* setup a rank_info object for us */
 489         pmix_globals.mypeer->info = PMIX_NEW(pmix_rank_info_t);
 490         if (NULL == pmix_globals.mypeer->info) {
 491             if (gdsfound) {
 492                 PMIX_INFO_DESTRUCT(&ginfo);
 493             }
 494             PMIX_RELEASE_THREAD(&pmix_global_lock);
 495             return PMIX_ERR_NOMEM;
 496         }
 497         pmix_globals.mypeer->info->pname.nspace = strdup(pmix_globals.myid.nspace);
 498         pmix_globals.mypeer->info->pname.rank = pmix_globals.myid.rank;
 499         /* our bfrops module will be set when we connect to the server */
 500     } else {
 501         /* select our bfrops compat module */
 502         pmix_globals.mypeer->nptr->compat.bfrops = pmix_bfrops_base_assign_module(NULL);
 503         if (NULL == pmix_globals.mypeer->nptr->compat.bfrops) {
 504             if (gdsfound) {
 505                 PMIX_INFO_DESTRUCT(&ginfo);
 506             }
 507             PMIX_RELEASE_THREAD(&pmix_global_lock);
 508             return PMIX_ERR_INIT;
 509         }
 510         /* the server will be using the same */
 511         pmix_client_globals.myserver->nptr->compat.bfrops = pmix_globals.mypeer->nptr->compat.bfrops;
 512     }
 513 
 514     /* select our psec compat module - the selection may be based
 515      * on the corresponding envars that should have been passed
 516      * to us at launch */
 517     evar = getenv("PMIX_SECURITY_MODE");
 518     pmix_globals.mypeer->nptr->compat.psec = pmix_psec_base_assign_module(evar);
 519     if (NULL == pmix_globals.mypeer->nptr->compat.psec) {
 520         if (gdsfound) {
 521             PMIX_INFO_DESTRUCT(&ginfo);
 522         }
 523         PMIX_RELEASE_THREAD(&pmix_global_lock);
 524         return PMIX_ERR_INIT;
 525     }
 526     /* the server will be using the same */
 527     pmix_client_globals.myserver->nptr->compat.psec = pmix_globals.mypeer->nptr->compat.psec;
 528 
 529     /* set the buffer type - the selection will be based
 530      * on the corresponding envars that should have been passed
 531      * to us at launch */
 532     evar = getenv("PMIX_BFROP_BUFFER_TYPE");
 533     if (NULL == evar) {
 534         /* just set to our default */
 535         pmix_globals.mypeer->nptr->compat.type = pmix_bfrops_globals.default_type;
 536     } else if (0 == strcmp(evar, "PMIX_BFROP_BUFFER_FULLY_DESC")) {
 537         pmix_globals.mypeer->nptr->compat.type = PMIX_BFROP_BUFFER_FULLY_DESC;
 538     } else {
 539         pmix_globals.mypeer->nptr->compat.type = PMIX_BFROP_BUFFER_NON_DESC;
 540     }
 541     /* the server will be using the same */
 542     pmix_client_globals.myserver->nptr->compat.type = pmix_globals.mypeer->nptr->compat.type;
 543 
 544     /* select a GDS module for our own internal use - the user may
 545      * have passed down a directive for this purpose. If they did, then
 546      * use it. Otherwise, we want the "hash" module */
 547     if (!gdsfound) {
 548         PMIX_INFO_LOAD(&ginfo, PMIX_GDS_MODULE, "hash", PMIX_STRING);
 549     }
 550     pmix_globals.mypeer->nptr->compat.gds = pmix_gds_base_assign_module(&ginfo, 1);
 551     if (NULL == pmix_globals.mypeer->nptr->compat.gds) {
 552         PMIX_INFO_DESTRUCT(&ginfo);
 553         PMIX_RELEASE_THREAD(&pmix_global_lock);
 554         return PMIX_ERR_INIT;
 555     }
 556     PMIX_INFO_DESTRUCT(&ginfo);
 557     /* select the gds compat module we will use to interact with
 558      * our server- the selection will be based
 559      * on the corresponding envars that should have been passed
 560      * to us at launch */
 561     evar = getenv("PMIX_GDS_MODULE");
 562     if (NULL != evar) {
 563         PMIX_INFO_LOAD(&ginfo, PMIX_GDS_MODULE, evar, PMIX_STRING);
 564         pmix_client_globals.myserver->nptr->compat.gds = pmix_gds_base_assign_module(&ginfo, 1);
 565         PMIX_INFO_DESTRUCT(&ginfo);
 566     } else {
 567         pmix_client_globals.myserver->nptr->compat.gds = pmix_gds_base_assign_module(NULL, 0);
 568     }
 569     if (NULL == pmix_client_globals.myserver->nptr->compat.gds) {
 570         PMIX_RELEASE_THREAD(&pmix_global_lock);
 571         return PMIX_ERR_INIT;
 572     }
 573 
 574     if (do_not_connect) {
 575         /* ensure we mark that we are not connected */
 576         pmix_globals.connected = false;
 577         /* it is an error if we were not given an nspace/rank */
 578         if (!nspace_given || !rank_given) {
 579             PMIX_RELEASE_THREAD(&pmix_global_lock);
 580             return PMIX_ERR_INIT;
 581         }
 582     } else {
 583         /* connect to the server */
 584         rc = pmix_ptl_base_connect_to_peer((struct pmix_peer_t*)pmix_client_globals.myserver, info, ninfo);
 585         if (PMIX_SUCCESS != rc){
 586             PMIX_RELEASE_THREAD(&pmix_global_lock);
 587             return rc;
 588         }
 589     }
 590     if (!nspace_given) {
 591         /* Success, so copy the nspace and rank to the proc struct they gave us */
 592         pmix_strncpy(proc->nspace, pmix_globals.myid.nspace, PMIX_MAX_NSLEN);
 593     }
 594     if (!rank_given) {
 595         proc->rank = pmix_globals.myid.rank;
 596     }
 597     /* and into our own peer object */
 598     if (NULL == pmix_globals.mypeer->nptr->nspace) {
 599         pmix_globals.mypeer->nptr->nspace = strdup(pmix_globals.myid.nspace);
 600     }
 601     /* setup a rank_info object for us */
 602     pmix_globals.mypeer->info = PMIX_NEW(pmix_rank_info_t);
 603     if (NULL == pmix_globals.mypeer->info) {
 604         PMIX_RELEASE_THREAD(&pmix_global_lock);
 605         return PMIX_ERR_NOMEM;
 606     }
 607     pmix_globals.mypeer->info->pname.nspace = strdup(pmix_globals.myid.nspace);
 608     pmix_globals.mypeer->info->pname.rank = pmix_globals.myid.rank;
 609 
 610     /* if we are acting as a server, then start listening */
 611     if (PMIX_PROC_IS_LAUNCHER(pmix_globals.mypeer)) {
 612         /* setup the wildcard recv for inbound messages from clients */
 613         rcv = PMIX_NEW(pmix_ptl_posted_recv_t);
 614         rcv->tag = UINT32_MAX;
 615         rcv->cbfunc = pmix_server_message_handler;
 616         /* add it to the end of the list of recvs */
 617         pmix_list_append(&pmix_ptl_globals.posted_recvs, &rcv->super);
 618         /* open the pnet framework so we can harvest envars */
 619         rc = pmix_mca_base_framework_open(&pmix_pnet_base_framework, 0);
 620         if (PMIX_SUCCESS != rc){
 621             PMIX_RELEASE_THREAD(&pmix_global_lock);
 622             return rc;
 623         }
 624         /* note that we do not select active plugins as we don't need them */
 625     }
 626 
 627     /* setup IOF */
 628     PMIX_IOF_SINK_DEFINE(&pmix_client_globals.iof_stdout, &pmix_globals.myid,
 629                          1, PMIX_FWD_STDOUT_CHANNEL, pmix_iof_write_handler);
 630     PMIX_IOF_SINK_DEFINE(&pmix_client_globals.iof_stderr, &pmix_globals.myid,
 631                          2, PMIX_FWD_STDERR_CHANNEL, pmix_iof_write_handler);
 632     if (fwd_stdin) {
 633         /* setup the read - we don't want to set nonblocking on our
 634          * stdio stream.  If we do so, we set the file descriptor to
 635          * non-blocking for everyone that has that file descriptor, which
 636          * includes everyone else in our shell pipeline chain.  (See
 637          * http://lists.freebsd.org/pipermail/freebsd-hackers/2005-January/009742.html).
 638          * This causes things like "prun -np 1 big_app | cat" to lose
 639          * output, because cat's stdout is then ALSO non-blocking and cat
 640          * isn't built to deal with that case (same with almost all other
 641          * unix text utils).*/
 642         fd = fileno(stdin);
 643         if (isatty(fd)) {
 644             /* We should avoid trying to read from stdin if we
 645              * have a terminal, but are backgrounded.  Catch the
 646              * signals that are commonly used when we switch
 647              * between being backgrounded and not.  If the
 648              * filedescriptor is not a tty, don't worry about it
 649              * and always stay connected.
 650              */
 651             pmix_event_signal_set(pmix_globals.evbase, &stdinsig,
 652                                   SIGCONT, pmix_iof_stdin_cb,
 653                                   &stdinev);
 654 
 655             /* setup a read event to read stdin, but don't activate it yet. The
 656              * dst_name indicates who should receive the stdin. If that recipient
 657              * doesn't do a corresponding pull, however, then the stdin will
 658              * be dropped upon receipt at the local daemon
 659              */
 660             PMIX_CONSTRUCT(&stdinev, pmix_iof_read_event_t);
 661             stdinev.fd = fd;
 662             stdinev.always_readable = pmix_iof_fd_always_ready(fd);
 663             if (stdinev.always_readable) {
 664                 pmix_event_evtimer_set(pmix_globals.evbase,
 665                                        &stdinev.ev,
 666                                        pmix_iof_read_local_handler,
 667                                        &stdinev);
 668             } else {
 669                 pmix_event_set(pmix_globals.evbase,
 670                                &stdinev.ev, fd,
 671                                PMIX_EV_READ,
 672                                pmix_iof_read_local_handler, &stdinev);
 673             }
 674             /* check to see if we want the stdin read event to be
 675              * active - we will always at least define the event,
 676              * but may delay its activation
 677              */
 678             if (pmix_iof_stdin_check(fd)) {
 679                 PMIX_IOF_READ_ACTIVATE(&stdinev);
 680             }
 681         } else {
 682                 /* if we are not looking at a tty, just setup a read event
 683                  * and activate it
 684                  */
 685             PMIX_CONSTRUCT(&stdinev, pmix_iof_read_event_t);
 686             stdinev.fd = fd;
 687             stdinev.always_readable = pmix_iof_fd_always_ready(fd);
 688             if (stdinev.always_readable) {
 689                 pmix_event_evtimer_set(pmix_globals.evbase,
 690                                        &stdinev.ev,
 691                                        pmix_iof_read_local_handler,
 692                                        &stdinev);
 693             } else {
 694                 pmix_event_set(pmix_globals.evbase,
 695                                &stdinev.ev, fd,
 696                                PMIX_EV_READ,
 697                                pmix_iof_read_local_handler, &stdinev);
 698             }                                                               \
 699             PMIX_IOF_READ_ACTIVATE(&stdinev);
 700         }
 701     }
 702 
 703     /* increment our init reference counter */
 704     pmix_globals.init_cntr++;
 705 
 706     /* if we are acting as a client, then send a request for our
 707      * job info - we do this as a non-blocking
 708      * transaction because some systems cannot handle very large
 709      * blocking operations and error out if we try them. */
 710     if (PMIX_PROC_IS_CLIENT(pmix_globals.mypeer)) {
 711          req = PMIX_NEW(pmix_buffer_t);
 712          PMIX_BFROPS_PACK(rc, pmix_client_globals.myserver,
 713                           req, &cmd, 1, PMIX_COMMAND);
 714          if (PMIX_SUCCESS != rc) {
 715             PMIX_ERROR_LOG(rc);
 716             PMIX_RELEASE(req);
 717             PMIX_RELEASE_THREAD(&pmix_global_lock);
 718             return rc;
 719         }
 720         /* send to the server */
 721         PMIX_CONSTRUCT(&cb, pmix_cb_t);
 722         PMIX_PTL_SEND_RECV(rc, pmix_client_globals.myserver,
 723                            req, job_data, (void*)&cb);
 724         if (PMIX_SUCCESS != rc) {
 725             PMIX_RELEASE_THREAD(&pmix_global_lock);
 726             return rc;
 727         }
 728         /* wait for the data to return */
 729         PMIX_WAIT_THREAD(&cb.lock);
 730         rc = cb.status;
 731         PMIX_DESTRUCT(&cb);
 732         if (PMIX_SUCCESS != rc) {
 733             PMIX_RELEASE_THREAD(&pmix_global_lock);
 734             return rc;
 735         }
 736         /* quick check to see if we got something back. If this
 737          * is a launcher that is being executed multiple times
 738          * in a job-script, then the original registration data
 739          * will have been deleted after the first invocation. In
 740          * such a case, we simply regenerate it locally as it is
 741          * well-known */
 742         pmix_cb_t cb;
 743         PMIX_CONSTRUCT(&cb, pmix_cb_t);
 744         pmix_strncpy(wildcard.nspace, pmix_globals.myid.nspace, PMIX_MAX_NSLEN);
 745         wildcard.rank = PMIX_RANK_WILDCARD;
 746         cb.proc = &wildcard;
 747         cb.copy = true;
 748         PMIX_GDS_FETCH_KV(rc, pmix_globals.mypeer, &cb);
 749         if (PMIX_SUCCESS != rc) {
 750             pmix_output_verbose(5, pmix_client_globals.get_output,
 751                                 "pmix:tool:client data not found in internal storage");
 752             rc = pmix_tool_init_info();
 753             if (PMIX_SUCCESS != rc) {
 754                 PMIX_RELEASE_THREAD(&pmix_global_lock);
 755                 return rc;
 756             }
 757         }
 758     } else {
 759         /* now finish the initialization by filling our local
 760          * datastore with typical job-related info. No point
 761          * in having the server generate these as we are
 762          * obviously a singleton, and so the values are well-known */
 763         rc = pmix_tool_init_info();
 764         if (PMIX_SUCCESS != rc) {
 765             PMIX_RELEASE_THREAD(&pmix_global_lock);
 766             return rc;
 767         }
 768     }
 769     PMIX_RELEASE_THREAD(&pmix_global_lock);
 770 
 771     /* if we are acting as a server, then start listening */
 772     if (PMIX_PROC_IS_LAUNCHER(pmix_globals.mypeer)) {
 773         /* start listening for connections */
 774         if (PMIX_SUCCESS != pmix_ptl_base_start_listening(info, ninfo)) {
 775             pmix_show_help("help-pmix-server.txt", "listener-thread-start", true);
 776             return PMIX_ERR_INIT;
 777         }
 778     }
 779 
 780     /* register the tool supported attrs */
 781     rc = pmix_register_tool_attrs();
 782     return rc;
 783 }
 784 
 785 pmix_status_t pmix_tool_init_info(void)
 786 {
 787     pmix_kval_t *kptr;
 788     pmix_status_t rc;
 789     pmix_proc_t wildcard;
 790     char hostname[PMIX_MAX_NSLEN];
 791 
 792     pmix_strncpy(wildcard.nspace, pmix_globals.myid.nspace, PMIX_MAX_NSLEN);
 793     wildcard.rank = pmix_globals.myid.rank;
 794 
 795     /* the jobid is just our nspace */
 796     kptr = PMIX_NEW(pmix_kval_t);
 797     kptr->key = strdup(PMIX_JOBID);
 798     PMIX_VALUE_CREATE(kptr->value, 1);
 799     kptr->value->type = PMIX_STRING;
 800     kptr->value->data.string = strdup(pmix_globals.myid.nspace);
 801     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
 802                       &wildcard,
 803                       PMIX_INTERNAL, kptr);
 804     if (PMIX_SUCCESS != rc) {
 805         PMIX_ERROR_LOG(rc);
 806         return rc;
 807     }
 808     PMIX_RELEASE(kptr); // maintain accounting
 809 
 810     /* our rank */
 811     kptr = PMIX_NEW(pmix_kval_t);
 812     kptr->key = strdup(PMIX_RANK);
 813     PMIX_VALUE_CREATE(kptr->value, 1);
 814     kptr->value->type = PMIX_INT;
 815     kptr->value->data.integer = 0;
 816     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
 817                       &pmix_globals.myid,
 818                       PMIX_INTERNAL, kptr);
 819     if (PMIX_SUCCESS != rc) {
 820         PMIX_ERROR_LOG(rc);
 821         return rc;
 822     }
 823     PMIX_RELEASE(kptr); // maintain accounting
 824 
 825     /* nproc offset */
 826     kptr = PMIX_NEW(pmix_kval_t);
 827     kptr->key = strdup(PMIX_NPROC_OFFSET);
 828     PMIX_VALUE_CREATE(kptr->value, 1);
 829     kptr->value->type = PMIX_UINT32;
 830     kptr->value->data.uint32 = 0;
 831     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
 832                       &wildcard,
 833                       PMIX_INTERNAL, kptr);
 834     if (PMIX_SUCCESS != rc) {
 835         PMIX_ERROR_LOG(rc);
 836         return rc;
 837     }
 838     PMIX_RELEASE(kptr); // maintain accounting
 839 
 840     /* node size */
 841     kptr = PMIX_NEW(pmix_kval_t);
 842     kptr->key = strdup(PMIX_NODE_SIZE);
 843     PMIX_VALUE_CREATE(kptr->value, 1);
 844     kptr->value->type = PMIX_UINT32;
 845     kptr->value->data.uint32 = 1;
 846     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
 847                       &wildcard,
 848                       PMIX_INTERNAL, kptr);
 849     if (PMIX_SUCCESS != rc) {
 850         PMIX_ERROR_LOG(rc);
 851         return rc;
 852     }
 853     PMIX_RELEASE(kptr); // maintain accounting
 854 
 855     /* local peers */
 856     kptr = PMIX_NEW(pmix_kval_t);
 857     kptr->key = strdup(PMIX_LOCAL_PEERS);
 858     PMIX_VALUE_CREATE(kptr->value, 1);
 859     kptr->value->type = PMIX_STRING;
 860     kptr->value->data.string = strdup("0");
 861     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
 862                       &wildcard,
 863                       PMIX_INTERNAL, kptr);
 864     if (PMIX_SUCCESS != rc) {
 865         PMIX_ERROR_LOG(rc);
 866         return rc;
 867     }
 868     PMIX_RELEASE(kptr); // maintain accounting
 869 
 870     /* local leader */
 871     kptr = PMIX_NEW(pmix_kval_t);
 872     kptr->key = strdup(PMIX_LOCALLDR);
 873     PMIX_VALUE_CREATE(kptr->value, 1);
 874     kptr->value->type = PMIX_UINT32;
 875     kptr->value->data.uint32 = 0;
 876     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
 877                       &wildcard,
 878                       PMIX_INTERNAL, kptr);
 879     if (PMIX_SUCCESS != rc) {
 880         PMIX_ERROR_LOG(rc);
 881         return rc;
 882     }
 883     PMIX_RELEASE(kptr); // maintain accounting
 884 
 885     /* universe size */
 886     kptr = PMIX_NEW(pmix_kval_t);
 887     kptr->key = strdup(PMIX_UNIV_SIZE);
 888     PMIX_VALUE_CREATE(kptr->value, 1);
 889     kptr->value->type = PMIX_UINT32;
 890     kptr->value->data.uint32 = 1;
 891     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
 892                       &wildcard,
 893                       PMIX_INTERNAL, kptr);
 894     if (PMIX_SUCCESS != rc) {
 895         PMIX_ERROR_LOG(rc);
 896         return rc;
 897     }
 898     PMIX_RELEASE(kptr); // maintain accounting
 899 
 900     /* job size - we are our very own job, so we have no peers */
 901     kptr = PMIX_NEW(pmix_kval_t);
 902     kptr->key = strdup(PMIX_JOB_SIZE);
 903     PMIX_VALUE_CREATE(kptr->value, 1);
 904     kptr->value->type = PMIX_UINT32;
 905     kptr->value->data.uint32 = 1;
 906     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
 907                       &wildcard,
 908                       PMIX_INTERNAL, kptr);
 909     if (PMIX_SUCCESS != rc) {
 910         PMIX_ERROR_LOG(rc);
 911         return rc;
 912     }
 913     PMIX_RELEASE(kptr); // maintain accounting
 914 
 915     /* local size - only us in our job */
 916     kptr = PMIX_NEW(pmix_kval_t);
 917     kptr->key = strdup(PMIX_LOCAL_SIZE);
 918     PMIX_VALUE_CREATE(kptr->value, 1);
 919     kptr->value->type = PMIX_UINT32;
 920     kptr->value->data.uint32 = 1;
 921     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
 922                       &wildcard,
 923                       PMIX_INTERNAL, kptr);
 924     if (PMIX_SUCCESS != rc) {
 925         PMIX_ERROR_LOG(rc);
 926         return rc;
 927     }
 928     PMIX_RELEASE(kptr); // maintain accounting
 929 
 930     /* max procs - since we are a self-started tool, there is no
 931      * allocation within which we can grow ourselves */
 932     kptr = PMIX_NEW(pmix_kval_t);
 933     kptr->key = strdup(PMIX_MAX_PROCS);
 934     PMIX_VALUE_CREATE(kptr->value, 1);
 935     kptr->value->type = PMIX_UINT32;
 936     kptr->value->data.uint32 = 1;
 937     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
 938                       &wildcard,
 939                       PMIX_INTERNAL, kptr);
 940     if (PMIX_SUCCESS != rc) {
 941         PMIX_ERROR_LOG(rc);
 942         return rc;
 943     }
 944     PMIX_RELEASE(kptr); // maintain accounting
 945 
 946     /* app number */
 947     kptr = PMIX_NEW(pmix_kval_t);
 948     kptr->key = strdup(PMIX_APPNUM);
 949     PMIX_VALUE_CREATE(kptr->value, 1);
 950     kptr->value->type = PMIX_UINT32;
 951     kptr->value->data.uint32 = 0;
 952     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
 953                       &pmix_globals.myid,
 954                       PMIX_INTERNAL, kptr);
 955     if (PMIX_SUCCESS != rc) {
 956         PMIX_ERROR_LOG(rc);
 957         return rc;
 958     }
 959     PMIX_RELEASE(kptr); // maintain accounting
 960 
 961     /* app leader */
 962     kptr = PMIX_NEW(pmix_kval_t);
 963     kptr->key = strdup(PMIX_APPLDR);
 964     PMIX_VALUE_CREATE(kptr->value, 1);
 965     kptr->value->type = PMIX_UINT32;
 966     kptr->value->data.uint32 = 0;
 967     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
 968                       &pmix_globals.myid,
 969                       PMIX_INTERNAL, kptr);
 970     if (PMIX_SUCCESS != rc) {
 971         PMIX_ERROR_LOG(rc);
 972         return rc;
 973     }
 974     PMIX_RELEASE(kptr); // maintain accounting
 975 
 976     /* app rank */
 977     kptr = PMIX_NEW(pmix_kval_t);
 978     kptr->key = strdup(PMIX_APP_RANK);
 979     PMIX_VALUE_CREATE(kptr->value, 1);
 980     kptr->value->type = PMIX_UINT32;
 981     kptr->value->data.uint32 = 0;
 982     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
 983                       &pmix_globals.myid,
 984                       PMIX_INTERNAL, kptr);
 985     if (PMIX_SUCCESS != rc) {
 986         PMIX_ERROR_LOG(rc);
 987         return rc;
 988     }
 989     PMIX_RELEASE(kptr); // maintain accounting
 990 
 991     /* global rank */
 992     kptr = PMIX_NEW(pmix_kval_t);
 993     kptr->key = strdup(PMIX_GLOBAL_RANK);
 994     PMIX_VALUE_CREATE(kptr->value, 1);
 995     kptr->value->type = PMIX_UINT32;
 996     kptr->value->data.uint32 = 0;
 997     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
 998                       &pmix_globals.myid,
 999                       PMIX_INTERNAL, kptr);
1000     if (PMIX_SUCCESS != rc) {
1001         PMIX_ERROR_LOG(rc);
1002         return rc;
1003     }
1004     PMIX_RELEASE(kptr); // maintain accounting
1005 
1006     /* local rank - we are alone in our job */
1007     kptr = PMIX_NEW(pmix_kval_t);
1008     kptr->key = strdup(PMIX_LOCAL_RANK);
1009     PMIX_VALUE_CREATE(kptr->value, 1);
1010     kptr->value->type = PMIX_UINT16;
1011     kptr->value->data.uint32 = 0;
1012     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
1013                       &pmix_globals.myid,
1014                       PMIX_INTERNAL, kptr);
1015     if (PMIX_SUCCESS != rc) {
1016         PMIX_ERROR_LOG(rc);
1017         return rc;
1018     }
1019     PMIX_RELEASE(kptr); // maintain accounting
1020 
1021     /* we cannot know the node rank as we don't know what
1022      * other processes are executing on this node - so
1023      * we'll add that info to the server-tool handshake
1024      * and load it from there */
1025 
1026     /* hostname */
1027     if (NULL != pmix_globals.hostname) {
1028         pmix_strncpy(hostname, pmix_globals.hostname, PMIX_MAX_NSLEN);
1029     } else {
1030         gethostname(hostname, PMIX_MAX_NSLEN);
1031     }
1032     kptr = PMIX_NEW(pmix_kval_t);
1033     kptr->key = strdup(PMIX_HOSTNAME);
1034     PMIX_VALUE_CREATE(kptr->value, 1);
1035     kptr->value->type = PMIX_STRING;
1036     kptr->value->data.string = strdup(hostname);
1037     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
1038                       &pmix_globals.myid,
1039                       PMIX_INTERNAL, kptr);
1040     if (PMIX_SUCCESS != rc) {
1041         PMIX_ERROR_LOG(rc);
1042         return rc;
1043     }
1044     PMIX_RELEASE(kptr); // maintain accounting
1045 
1046     /* we cannot know the RM's nodeid for this host, so
1047      * we'll add that info to the server-tool handshake
1048      * and load it from there */
1049 
1050     /* the nodemap is simply our hostname as there is no
1051      * regex to generate */
1052     kptr = PMIX_NEW(pmix_kval_t);
1053     kptr->key = strdup(PMIX_NODE_MAP);
1054     PMIX_VALUE_CREATE(kptr->value, 1);
1055     kptr->value->type = PMIX_STRING;
1056     kptr->value->data.string = strdup(hostname);
1057     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
1058                       &wildcard,
1059                       PMIX_INTERNAL, kptr);
1060     if (PMIX_SUCCESS != rc) {
1061         PMIX_ERROR_LOG(rc);
1062         return rc;
1063     }
1064     PMIX_RELEASE(kptr); // maintain accounting
1065 
1066     /* likewise, the proc map is just our rank as we are
1067      * the only proc in this job */
1068     kptr = PMIX_NEW(pmix_kval_t);
1069     kptr->key = strdup(PMIX_PROC_MAP);
1070     PMIX_VALUE_CREATE(kptr->value, 1);
1071     kptr->value->type = PMIX_STRING;
1072     kptr->value->data.string = strdup("0");
1073     PMIX_GDS_STORE_KV(rc, pmix_globals.mypeer,
1074                       &wildcard,
1075                       PMIX_INTERNAL, kptr);
1076     if (PMIX_SUCCESS != rc) {
1077         PMIX_ERROR_LOG(rc);
1078         return rc;
1079     }
1080     PMIX_RELEASE(kptr); // maintain accounting
1081 
1082     return PMIX_SUCCESS;
1083 }
1084 
1085 
1086 typedef struct {
1087     pmix_lock_t lock;
1088     pmix_event_t ev;
1089     bool active;
1090 } pmix_tool_timeout_t;
1091 
1092 /* timer callback */
1093 static void fin_timeout(int sd, short args, void *cbdata)
1094 {
1095     pmix_tool_timeout_t *tev;
1096     tev = (pmix_tool_timeout_t*)cbdata;
1097 
1098     pmix_output_verbose(2, pmix_globals.debug_output,
1099                         "pmix:tool finwait timeout fired");
1100     if (tev->active) {
1101         tev->active = false;
1102         PMIX_WAKEUP_THREAD(&tev->lock);
1103     }
1104 }
1105 /* callback for finalize completion */
1106 static void finwait_cbfunc(struct pmix_peer_t *pr,
1107                            pmix_ptl_hdr_t *hdr,
1108                            pmix_buffer_t *buf, void *cbdata)
1109 {
1110     pmix_tool_timeout_t *tev;
1111     tev = (pmix_tool_timeout_t*)cbdata;
1112 
1113     pmix_output_verbose(2, pmix_globals.debug_output,
1114                         "pmix:tool finwait_cbfunc received");
1115     if (tev->active) {
1116         tev->active = false;
1117         pmix_event_del(&tev->ev);  // stop the timer
1118     }
1119     PMIX_WAKEUP_THREAD(&tev->lock);
1120 }
1121 
1122 PMIX_EXPORT pmix_status_t PMIx_tool_finalize(void)
1123 {
1124     pmix_buffer_t *msg;
1125     pmix_cmd_t cmd = PMIX_FINALIZE_CMD;
1126     pmix_status_t rc;
1127     pmix_tool_timeout_t tev;
1128     struct timeval tv = {5, 0};
1129     int n;
1130     pmix_peer_t *peer;
1131 
1132     PMIX_ACQUIRE_THREAD(&pmix_global_lock);
1133     if (1 != pmix_globals.init_cntr) {
1134         --pmix_globals.init_cntr;
1135         PMIX_RELEASE_THREAD(&pmix_global_lock);
1136         return PMIX_SUCCESS;
1137     }
1138     pmix_globals.init_cntr = 0;
1139     pmix_globals.mypeer->finalized = true;
1140     PMIX_RELEASE_THREAD(&pmix_global_lock);
1141 
1142     pmix_output_verbose(2, pmix_globals.debug_output,
1143                         "pmix:tool finalize called");
1144 
1145     /* flush anything that is still trying to be written out */
1146     pmix_iof_static_dump_output(&pmix_client_globals.iof_stdout);
1147     pmix_iof_static_dump_output(&pmix_client_globals.iof_stderr);
1148     PMIX_DESTRUCT(&pmix_client_globals.iof_stdout);
1149     PMIX_DESTRUCT(&pmix_client_globals.iof_stderr);
1150 
1151     /* if we are connected, then disconnect */
1152     if (pmix_globals.connected) {
1153         pmix_output_verbose(2, pmix_globals.debug_output,
1154                             "pmix:tool sending finalize sync to server");
1155 
1156         /* setup a cmd message to notify the PMIx
1157          * server that we are normally terminating */
1158         msg = PMIX_NEW(pmix_buffer_t);
1159         /* pack the cmd */
1160         PMIX_BFROPS_PACK(rc, pmix_client_globals.myserver,
1161                          msg, &cmd, 1, PMIX_COMMAND);
1162         if (PMIX_SUCCESS != rc) {
1163             PMIX_ERROR_LOG(rc);
1164             PMIX_RELEASE(msg);
1165             return rc;
1166         }
1167         /* setup a timer to protect ourselves should the server be unable
1168          * to answer for some reason */
1169         PMIX_CONSTRUCT_LOCK(&tev.lock);
1170         pmix_event_assign(&tev.ev, pmix_globals.evbase, -1, 0,
1171                           fin_timeout, &tev);
1172         tev.active = true;
1173         PMIX_POST_OBJECT(&tev);
1174         pmix_event_add(&tev.ev, &tv);
1175         PMIX_PTL_SEND_RECV(rc, pmix_client_globals.myserver, msg,
1176                            finwait_cbfunc, (void*)&tev);
1177         if (PMIX_SUCCESS != rc) {
1178             if (tev.active) {
1179                 pmix_event_del(&tev.ev);
1180             }
1181             return rc;
1182         }
1183 
1184         /* wait for the ack to return */
1185         PMIX_WAIT_THREAD(&tev.lock);
1186         PMIX_DESTRUCT_LOCK(&tev.lock);
1187 
1188         if (tev.active) {
1189             pmix_event_del(&tev.ev);
1190         }
1191         pmix_output_verbose(2, pmix_globals.debug_output,
1192                             "pmix:tool finalize sync received");
1193 
1194     }
1195 
1196     if (!pmix_globals.external_evbase) {
1197         /* stop the progress thread, but leave the event base
1198          * still constructed. This will allow us to safely
1199          * tear down the infrastructure, including removal
1200          * of any events objects may be holding */
1201         (void)pmix_progress_thread_pause(NULL);
1202     }
1203 
1204 //    PMIX_RELEASE(pmix_client_globals.myserver);
1205     PMIX_LIST_DESTRUCT(&pmix_client_globals.pending_requests);
1206     for (n=0; n < pmix_client_globals.peers.size; n++) {
1207         if (NULL != (peer = (pmix_peer_t*)pmix_pointer_array_get_item(&pmix_client_globals.peers, n))) {
1208             PMIX_RELEASE(peer);
1209         }
1210     }
1211 
1212     if (PMIX_PROC_IS_LAUNCHER(pmix_globals.mypeer)) {
1213         pmix_ptl_base_stop_listening();
1214 
1215         for (n=0; n < pmix_server_globals.clients.size; n++) {
1216             if (NULL != (peer = (pmix_peer_t*)pmix_pointer_array_get_item(&pmix_server_globals.clients, n))) {
1217                 PMIX_RELEASE(peer);
1218             }
1219         }
1220 
1221         (void)pmix_mca_base_framework_close(&pmix_pnet_base_framework);
1222         PMIX_DESTRUCT(&pmix_server_globals.clients);
1223         PMIX_LIST_DESTRUCT(&pmix_server_globals.collectives);
1224         PMIX_LIST_DESTRUCT(&pmix_server_globals.remote_pnd);
1225         PMIX_LIST_DESTRUCT(&pmix_server_globals.local_reqs);
1226         PMIX_LIST_DESTRUCT(&pmix_server_globals.gdata);
1227         PMIX_LIST_DESTRUCT(&pmix_server_globals.events);
1228         PMIX_LIST_DESTRUCT(&pmix_server_globals.nspaces);
1229         PMIX_LIST_DESTRUCT(&pmix_server_globals.iof);
1230     }
1231 
1232     /* shutdown services */
1233     pmix_rte_finalize();
1234     if (NULL != pmix_globals.mypeer) {
1235         PMIX_RELEASE(pmix_globals.mypeer);
1236     }
1237 
1238     /* finalize the class/object system */
1239     pmix_class_finalize();
1240 
1241     return PMIX_SUCCESS;
1242 }
1243 
1244 pmix_status_t PMIx_tool_connect_to_server(pmix_proc_t *proc,
1245                                           pmix_info_t info[], size_t ninfo)
1246 {
1247     pmix_buffer_t *msg;
1248     pmix_cmd_t cmd = PMIX_FINALIZE_CMD;
1249     pmix_status_t rc;
1250     pmix_tool_timeout_t tev;
1251     struct timeval tv = {2, 0};
1252 
1253     PMIX_ACQUIRE_THREAD(&pmix_global_lock);
1254     if (pmix_globals.init_cntr <= 0) {
1255         PMIX_RELEASE_THREAD(&pmix_global_lock);
1256         return PMIX_ERR_INIT;
1257     }
1258     PMIX_RELEASE_THREAD(&pmix_global_lock);
1259 
1260     /* check for bozo error - we do this
1261      * prior to breaking any existing connection to avoid
1262      * becoming stranded due to an incorrect function call */
1263     if (NULL == info || 0 == ninfo) {
1264         pmix_show_help("help-pmix-runtime.txt", "tool:no-server", true);
1265         return PMIX_ERR_BAD_PARAM;
1266     }
1267 
1268     /* break any existing connection */
1269     if (pmix_globals.connected) {
1270         /* gracefully terminate this connection */
1271         msg = PMIX_NEW(pmix_buffer_t);
1272         /* pack the cmd */
1273         PMIX_BFROPS_PACK(rc, pmix_client_globals.myserver,
1274                          msg, &cmd, 1, PMIX_COMMAND);
1275         if (PMIX_SUCCESS != rc) {
1276             PMIX_ERROR_LOG(rc);
1277             PMIX_RELEASE(msg);
1278             return rc;
1279         }
1280 
1281 
1282         pmix_output_verbose(2, pmix_globals.debug_output,
1283                             "pmix:tool:reconnect sending finalize sync to server");
1284 
1285         /* setup a timer to protect ourselves should the server be unable
1286          * to answer for some reason */
1287         PMIX_CONSTRUCT_LOCK(&tev.lock);
1288         pmix_event_assign(&tev.ev, pmix_globals.evbase, -1, 0,
1289                           fin_timeout, &tev);
1290         tev.active = true;
1291         PMIX_POST_OBJECT(&tev);
1292         pmix_event_add(&tev.ev, &tv);
1293         PMIX_PTL_SEND_RECV(rc, pmix_client_globals.myserver, msg,
1294                            finwait_cbfunc, (void*)&tev);
1295         if (PMIX_SUCCESS != rc) {
1296             if (tev.active) {
1297                 pmix_event_del(&tev.ev);
1298             }
1299             return rc;
1300         }
1301 
1302         /* wait for the ack to return */
1303         PMIX_WAIT_THREAD(&tev.lock);
1304         PMIX_DESTRUCT_LOCK(&tev.lock);
1305         if (tev.active) {
1306             pmix_event_del(&tev.ev);
1307         }
1308         pmix_output_verbose(2, pmix_globals.debug_output,
1309                             "pmix:tool:reconnect finalize sync received");
1310     }
1311 
1312     /* now ask the ptl to establish connection to the new server */
1313     rc = pmix_ptl_base_connect_to_peer((struct pmix_peer_t*)pmix_client_globals.myserver, info, ninfo);
1314     return rc;
1315 }
/* [<][>][^][v][top][bottom][index][help] */
root/opal/mca/pmix/pmix4x/pmix/src/tool/pmix_tool.c

DEFINITIONS
