opal/mca/pmix/pmix4x/pmix/src/mca/ptl/usock/ptl_usock

/* [<][>][^][v][top][bottom][index][help] */
This source file includes following definitions.
component_open
component_close
component_query
setup_listener
listener_cb
connection_handler
   1 /* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
   2 /*
   3  * Copyright (c) 2004-2008 The Trustees of Indiana University and Indiana
   4  *                         University Research and Technology
   5  *                         Corporation.  All rights reserved.
   6  * Copyright (c) 2004-2005 The University of Tennptlee and The University
   7  *                         of Tennptlee Research Foundation.  All rights
   8  *                         reserved.
   9  * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
  10  *                         University of Stuttgart.  All rights reserved.
  11  * Copyright (c) 2004-2005 The Regents of the University of California.
  12  *                         All rights reserved.
  13  * Copyright (c) 2015      Los Alamos National Security, LLC. All rights
  14  *                         reserved.
  15  * Copyright (c) 2016-2018 Intel, Inc.  All rights reserved.
  16  * Copyright (c) 2017      Research Organization for Information Science
  17  *                         and Technology (RIST). All rights reserved.
  18  * Copyright (c) 2018      IBM Corporation.  All rights reserved.
  19  * $COPYRIGHT$
  20  *
  21  * Additional copyrights may follow
  22  *
  23  * $HEADER$
  24  *
  25  * These symbols are in a file by themselves to provide nice linker
  26  * semantics.  Since linkers generally pull in symbols by object
  27  * files, keeping these symbols as the only symbols in this file
  28  * prevents utility programs such as "ompi_info" from having to import
  29  * entire components just to query their version and parameters.
  30  */
  31 
  32 #include <src/include/pmix_config.h>
  33 #include <pmix_common.h>
  34 
  35 #ifdef HAVE_FCNTL_H
  36 #include <fcntl.h>
  37 #endif
  38 #ifdef HAVE_UNISTD_H
  39 #include <unistd.h>
  40 #endif
  41 #ifdef HAVE_SYS_SOCKET_H
  42 #include <sys/socket.h>
  43 #endif
  44 #ifdef HAVE_SYS_UN_H
  45 #include <sys/un.h>
  46 #endif
  47 #ifdef HAVE_SYS_UIO_H
  48 #include <sys/uio.h>
  49 #endif
  50 #ifdef HAVE_SYS_TYPES_H
  51 #include <sys/types.h>
  52 #endif
  53 #ifdef HAVE_SYS_STAT_H
  54 #include <sys/stat.h>
  55 #endif
  56 #include <errno.h>
  57 
  58 #include "src/util/argv.h"
  59 #include "src/util/error.h"
  60 #include "src/util/fd.h"
  61 #include "src/util/show_help.h"
  62 #include "src/util/strnlen.h"
  63 #include "src/mca/bfrops/base/base.h"
  64 #include "src/mca/gds/base/base.h"
  65 #include "src/mca/psec/base/base.h"
  66 #include "src/server/pmix_server_ops.h"
  67 
  68 #include "src/mca/ptl/base/base.h"
  69 #include "src/mca/ptl/usock/ptl_usock.h"
  70 
  71 static pmix_status_t component_open(void);
  72 static pmix_status_t component_close(void);
  73 static int component_query(pmix_mca_base_module_t **module, int *priority);
  74 static pmix_status_t setup_listener(pmix_info_t info[], size_t ninfo,
  75                                     bool *need_listener);
  76 
  77 /*
  78  * Instantiate the public struct with all of our public information
  79  * and pointers to our public functions in it
  80  */
  81 PMIX_EXPORT pmix_ptl_usock_component_t mca_ptl_usock_component = {
  82     .super = {
  83         .base = {
  84             PMIX_PTL_BASE_VERSION_1_0_0,
  85 
  86             /* Component name and version */
  87             .pmix_mca_component_name = "usock",
  88             PMIX_MCA_BASE_MAKE_VERSION(component,
  89                                        PMIX_MAJOR_VERSION,
  90                                        PMIX_MINOR_VERSION,
  91                                        PMIX_RELEASE_VERSION),
  92 
  93             /* Component open and close functions */
  94             .pmix_mca_open_component = component_open,
  95             .pmix_mca_close_component = component_close,
  96             .pmix_mca_query_component = component_query
  97         },
  98         .priority = 15,
  99         .uri = NULL,
 100         .setup_listener = setup_listener
 101     },
 102     .tmpdir = NULL,
 103     .filename = NULL
 104 };
 105 
 106 static void connection_handler(int sd, short args, void *cbdata);
 107 static void listener_cb(int incoming_sd, void *cbdata);
 108 
 109 pmix_status_t component_open(void)
 110 {
 111     char *tdir;
 112 
 113     memset(&mca_ptl_usock_component.connection, 0, sizeof(mca_ptl_usock_component.connection));
 114 
 115     /* check for environ-based directives
 116      * on system tmpdir to use */
 117     if (NULL == (tdir = getenv("PMIX_SYSTEM_TMPDIR"))) {
 118         if (NULL == (tdir = getenv("TMPDIR"))) {
 119             if (NULL == (tdir = getenv("TEMP"))) {
 120                 if (NULL == (tdir = getenv("TMP"))) {
 121                     tdir = "/tmp";
 122                 }
 123             }
 124         }
 125     }
 126     if (NULL != tdir) {
 127         mca_ptl_usock_component.tmpdir = strdup(tdir);
 128     }
 129 
 130     return PMIX_SUCCESS;
 131 }
 132 
 133 
 134 pmix_status_t component_close(void)
 135 {
 136     if (NULL != mca_ptl_usock_component.tmpdir) {
 137         free(mca_ptl_usock_component.tmpdir);
 138     }
 139     if (NULL != mca_ptl_usock_component.super.uri) {
 140         free(mca_ptl_usock_component.super.uri);
 141     }
 142     if (NULL != mca_ptl_usock_component.filename) {
 143         /* remove the file */
 144         unlink(mca_ptl_usock_component.filename);
 145         free(mca_ptl_usock_component.filename);
 146     }
 147 
 148     return PMIX_SUCCESS;
 149 }
 150 
 151 static int component_query(pmix_mca_base_module_t **module, int *priority)
 152 {
 153     *module = (pmix_mca_base_module_t*)&pmix_ptl_usock_module;
 154     return PMIX_SUCCESS;
 155 }
 156 
 157 /* if we are the server, then we need to setup a usock rendezvous
 158  * point for legacy releases, but only do so if requested as some
 159  * systems may not wish to support older releases. The system can,
 160  * of course, simply use the MCA param method to disable this
 161  * component (PMIX_MCA_ptl=^usock), or can tell us to disqualify
 162  * ourselves using an info key to this API.
 163  *
 164  * NOTE: we accept MCA parameters, but info keys override them
 165  */
 166 static pmix_status_t setup_listener(pmix_info_t info[], size_t ninfo,
 167                                     bool *need_listener)
 168 {
 169     int flags;
 170     size_t n;
 171     pmix_listener_t *lt;
 172     pmix_status_t rc;
 173     socklen_t addrlen;
 174     struct sockaddr_un *address;
 175     bool disabled = false;
 176     char *pmix_pid;
 177     pid_t mypid;
 178 
 179     pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
 180                         "ptl:usock setup_listener");
 181 
 182     /* if we are not a server, then we shouldn't be doing this */
 183     if (!PMIX_PROC_IS_SERVER(pmix_globals.mypeer)) {
 184         return PMIX_ERR_NOT_SUPPORTED;
 185     }
 186 
 187     /* scan the info keys and process any override instructions */
 188     if (NULL != info) {
 189         for (n=0; n < ninfo; n++) {
 190             if (0 == strcmp(info[n].key, PMIX_USOCK_DISABLE)) {
 191                 disabled = PMIX_INFO_TRUE(&info[n]);;
 192                 break;
 193             }
 194         }
 195     }
 196 
 197     /* see if we have been disabled */
 198     if (disabled) {
 199         pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
 200                             "ptl:usock not available");
 201         return PMIX_ERR_NOT_AVAILABLE;
 202     }
 203 
 204     pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
 205                         "ptl:usock setting up listener");
 206 
 207     addrlen = sizeof(struct sockaddr_un);
 208     address = (struct sockaddr_un*)&mca_ptl_usock_component.connection;
 209     address->sun_family = AF_UNIX;
 210 
 211     /* define the listener */
 212     lt = PMIX_NEW(pmix_listener_t);
 213 
 214     /* for now, just setup the v1.1 series rendezvous point
 215      * we use the pid to reduce collisions */
 216     mypid = getpid();
 217     if (0 > asprintf(&pmix_pid, "%s/pmix-%d", mca_ptl_usock_component.tmpdir, mypid)) {
 218         PMIX_RELEASE(lt);
 219         return PMIX_ERR_NOMEM;
 220     }
 221     if ((strlen(pmix_pid) + 1) > sizeof(address->sun_path)-1) {
 222         pmix_show_help("help-pmix-server.txt", "rnd-path-too-long", true,
 223                        mca_ptl_usock_component.tmpdir, pmix_pid);
 224         free(pmix_pid);
 225         PMIX_RELEASE(lt);
 226         return PMIX_ERR_INVALID_LENGTH;
 227     }
 228     snprintf(address->sun_path, sizeof(address->sun_path)-1, "%s", pmix_pid);
 229     free(pmix_pid);
 230     /* set the URI */
 231     lt->varname = strdup("PMIX_SERVER_URI:PMIX_SERVER_URI2USOCK");
 232     if (0 > asprintf(&lt->uri, "%s:%lu:%s", pmix_globals.myid.nspace,
 233                     (unsigned long)pmix_globals.myid.rank, address->sun_path)) {
 234         PMIX_RELEASE(lt);
 235         return PMIX_ERR_NOMEM;
 236     }
 237     /* save the rendezvous filename for later removal */
 238     mca_ptl_usock_component.filename = strdup(address->sun_path);
 239 
 240     lt->protocol = PMIX_PROTOCOL_V1;
 241     lt->ptl = (struct pmix_ptl_module_t*)&pmix_ptl_usock_module;
 242     lt->cbfunc = connection_handler;
 243     pmix_list_append(&pmix_ptl_globals.listeners, &lt->super);
 244 
 245     /* create a listen socket for incoming connection attempts */
 246     lt->socket = socket(PF_UNIX, SOCK_STREAM, 0);
 247     if (lt->socket < 0) {
 248         printf("%s:%d socket() failed\n", __FILE__, __LINE__);
 249         goto sockerror;
 250     }
 251     /* Set the socket to close-on-exec so that no children inherit
 252      * this FD */
 253     if (pmix_fd_set_cloexec(lt->socket) != PMIX_SUCCESS) {
 254         CLOSE_THE_SOCKET(lt->socket);
 255         goto sockerror;
 256     }
 257 
 258     if (bind(lt->socket, (struct sockaddr*)address, addrlen) < 0) {
 259         printf("%s:%d bind() failed\n", __FILE__, __LINE__);
 260         CLOSE_THE_SOCKET(lt->socket);
 261         goto sockerror;
 262     }
 263     /* chown as required */
 264     if (lt->owner_given) {
 265         if (0 != chown(address->sun_path, lt->owner, -1)) {
 266             pmix_output(0, "CANNOT CHOWN socket %s: %s", address->sun_path, strerror (errno));
 267             CLOSE_THE_SOCKET(lt->socket);
 268             goto sockerror;
 269         }
 270     }
 271     if (lt->group_given) {
 272         if (0 != chown(address->sun_path, -1, lt->group)) {
 273             pmix_output(0, "CANNOT CHOWN socket %s: %s", address->sun_path, strerror (errno));
 274             CLOSE_THE_SOCKET(lt->socket);
 275             goto sockerror;
 276         }
 277     }
 278     /* set the mode as required */
 279     if (0 != chmod(address->sun_path, lt->mode)) {
 280         pmix_output(0, "CANNOT CHMOD socket %s: %s", address->sun_path, strerror (errno));
 281         CLOSE_THE_SOCKET(lt->socket);
 282         goto sockerror;
 283     }
 284 
 285     /* setup listen backlog to maximum allowed by kernel */
 286     if (listen(lt->socket, SOMAXCONN) < 0) {
 287         printf("%s:%d listen() failed\n", __FILE__, __LINE__);
 288         CLOSE_THE_SOCKET(lt->socket);
 289         goto sockerror;
 290     }
 291 
 292     /* set socket up to be non-blocking, otherwise accept could block */
 293     if ((flags = fcntl(lt->socket, F_GETFL, 0)) < 0) {
 294         printf("%s:%d fcntl(F_GETFL) failed\n", __FILE__, __LINE__);
 295         CLOSE_THE_SOCKET(lt->socket);
 296         goto sockerror;
 297     }
 298     flags |= O_NONBLOCK;
 299     if (fcntl(lt->socket, F_SETFL, flags) < 0) {
 300         printf("%s:%d fcntl(F_SETFL) failed\n", __FILE__, __LINE__);
 301         CLOSE_THE_SOCKET(lt->socket);
 302         goto sockerror;
 303     }
 304 
 305     /* if the server will listen for us, then ask it to do so now */
 306     rc = PMIX_ERR_NOT_SUPPORTED;
 307     if (NULL != pmix_host_server.listener) {
 308         rc = pmix_host_server.listener(lt->socket, listener_cb, (void*)lt);
 309     }
 310 
 311     if (PMIX_SUCCESS != rc) {
 312         *need_listener = true;
 313     }
 314 
 315     return PMIX_SUCCESS;
 316 
 317   sockerror:
 318       pmix_list_remove_item(&pmix_ptl_globals.listeners, &lt->super);
 319       PMIX_RELEASE(lt);
 320       return PMIX_ERROR;
 321 }
 322 
 323 static void listener_cb(int incoming_sd, void *cbdata)
 324 {
 325     pmix_pending_connection_t *pending_connection;
 326 
 327     /* throw it into our event library for processing */
 328     pmix_output_verbose(8, pmix_ptl_base_framework.framework_output,
 329                         "listen_cb: pushing new connection %d into evbase",
 330                         incoming_sd);
 331     pending_connection = PMIX_NEW(pmix_pending_connection_t);
 332     pending_connection->sd = incoming_sd;
 333     pmix_event_assign(&pending_connection->ev, pmix_globals.evbase, -1,
 334                       EV_WRITE, connection_handler, pending_connection);
 335     pmix_event_active(&pending_connection->ev, EV_WRITE, 1);
 336 }
 337 
 338 static void connection_handler(int sd, short args, void *cbdata)
 339 {
 340     pmix_pending_connection_t *pnd = (pmix_pending_connection_t*)cbdata;
 341     char *msg, *ptr, *nspace, *version, *sec, *bfrops, *gds;
 342     pmix_status_t rc;
 343     unsigned int rank;
 344     pmix_usock_hdr_t hdr;
 345     pmix_namespace_t *nptr, *tmp;
 346     pmix_rank_info_t *info;
 347     pmix_peer_t *psave = NULL;
 348     bool found;
 349     pmix_proc_t proc;
 350     size_t len;
 351     pmix_bfrop_buffer_type_t bftype;
 352     char **vers;
 353     int major, minor, rel;
 354     unsigned int msglen;
 355     pmix_info_t ginfo;
 356     pmix_byte_object_t cred;
 357 
 358     /* acquire the object */
 359     PMIX_ACQUIRE_OBJECT(pnd);
 360 
 361     pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
 362                         "USOCK CONNECTION FROM PEER ON SOCKET %d", pnd->sd);
 363 
 364     /* ensure all is zero'd */
 365     memset(&hdr, 0, sizeof(pmix_usock_hdr_t));
 366 
 367     /* get the header */
 368     if (PMIX_SUCCESS != (rc = pmix_ptl_base_recv_blocking(pnd->sd, (char*)&hdr, sizeof(pmix_usock_hdr_t)))) {
 369         CLOSE_THE_SOCKET(pnd->sd);
 370         PMIX_RELEASE(pnd);
 371         return;
 372     }
 373 
 374     /* get the id, authentication and version payload (and possibly
 375      * security credential) - to guard against potential attacks,
 376      * we'll set an arbitrary limit per a define */
 377     if (PMIX_MAX_CRED_SIZE < hdr.nbytes) {
 378         CLOSE_THE_SOCKET(pnd->sd);
 379         PMIX_RELEASE(pnd);
 380         return;
 381     }
 382     if (NULL == (msg = (char*)malloc(hdr.nbytes))) {
 383         CLOSE_THE_SOCKET(pnd->sd);
 384         PMIX_RELEASE(pnd);
 385         return;
 386     }
 387     if (PMIX_SUCCESS != pmix_ptl_base_recv_blocking(pnd->sd, msg, hdr.nbytes)) {
 388         /* unable to complete the recv */
 389         pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
 390                             "unable to complete recv of connect-ack with client ON SOCKET %d", pnd->sd);
 391         free(msg);
 392         CLOSE_THE_SOCKET(pnd->sd);
 393         PMIX_RELEASE(pnd);
 394         return;
 395     }
 396     len = hdr.nbytes;
 397     ptr = msg;
 398 
 399     /* extract the nspace of the requestor */
 400     PMIX_STRNLEN(msglen, ptr, len);
 401     if (msglen < len) {
 402         nspace = ptr;
 403         ptr += strlen(nspace) + 1;
 404         len -= strlen(nspace) + 1;
 405     } else {
 406         free(msg);
 407         CLOSE_THE_SOCKET(pnd->sd);
 408         PMIX_RELEASE(pnd);
 409         return;
 410     }
 411 
 412     /* extract the rank */
 413     PMIX_STRNLEN(msglen, ptr, len);
 414     if (msglen <= len) {
 415         memcpy(&rank, ptr, sizeof(int));
 416         ptr += sizeof(int);
 417         len -= sizeof(int);
 418     } else {
 419         free(msg);
 420         CLOSE_THE_SOCKET(pnd->sd);
 421         PMIX_RELEASE(pnd);
 422         return;
 423     }
 424 
 425     /* get their version string */
 426     PMIX_STRNLEN(msglen, ptr, len);
 427     if (msglen < len) {
 428         version = ptr;
 429         ptr += strlen(version) + 1;
 430         len -= strlen(version) + 1;
 431     } else {
 432         free(msg);
 433         CLOSE_THE_SOCKET(pnd->sd);
 434         PMIX_RELEASE(pnd);
 435         return;
 436     }
 437 
 438     /* check the version - we do NOT support anything less than
 439      * v1.2.5 */
 440     vers = pmix_argv_split(version, '.');
 441     major = strtol(vers[0], NULL, 10);
 442     minor = strtol(vers[1], NULL, 10);
 443     rel = strtol(vers[2], NULL, 10);
 444     pmix_argv_free(vers);
 445     if (1 == major && (2 != minor || 5 > rel)) {
 446         pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
 447                             "connection request from client of unsupported version %s", version);
 448         free(msg);
 449         CLOSE_THE_SOCKET(pnd->sd);
 450         PMIX_RELEASE(pnd);
 451         return;
 452     }
 453 
 454     /* get any provided credential */
 455     if (1 == major) {
 456         /* credential is always a string */
 457         PMIX_STRNLEN(msglen, ptr, len);
 458         if (msglen < len) {
 459             cred.bytes = ptr;
 460             cred.size = strlen(ptr);
 461             ptr += cred.size + 1;
 462             len -= cred.size + 1;
 463         } else {
 464             free(msg);
 465             CLOSE_THE_SOCKET(pnd->sd);
 466             PMIX_RELEASE(pnd);
 467             return;
 468         }
 469     } else {
 470         /* credential could be something else */
 471         if (sizeof(size_t) < len) {
 472             memcpy(&cred.size, ptr, sizeof(size_t));
 473             ptr += sizeof(size_t);
 474             len -= sizeof(size_t);
 475         } else {
 476             free(msg);
 477             CLOSE_THE_SOCKET(pnd->sd);
 478             PMIX_RELEASE(pnd);
 479             return;
 480         }
 481         if (0 < cred.size) {
 482             cred.bytes = ptr;
 483             ptr += cred.size;
 484             len -= cred.size;
 485         } else {
 486             /* set cred pointer to NULL to guard against validation
 487              * methods that assume a zero length credential is NULL */
 488             cred.bytes = NULL;
 489         }
 490     }
 491 
 492     /* get their sec module */
 493     PMIX_STRNLEN(msglen, ptr, len);
 494     if (msglen < len) {
 495         sec = ptr;
 496         ptr += strlen(sec) + 1;
 497         len -= strlen(sec) + 1;
 498     } else {
 499         free(msg);
 500         CLOSE_THE_SOCKET(pnd->sd);
 501         PMIX_RELEASE(pnd);
 502         return;
 503     }
 504 
 505     /* get their bfrops module */
 506     PMIX_STRNLEN(msglen, ptr, len);
 507     if (msglen < len) {
 508         bfrops = ptr;
 509         ptr += strlen(bfrops) + 1;
 510         len -= strlen(bfrops) + 1;
 511     } else {
 512         free(msg);
 513         CLOSE_THE_SOCKET(pnd->sd);
 514         PMIX_RELEASE(pnd);
 515         return;
 516     }
 517 
 518     /* get their buffer type */
 519     if (0 < len) {
 520         bftype = ptr[0];
 521         ptr += 1;
 522         len -= 1;
 523     } else {
 524         free(msg);
 525         CLOSE_THE_SOCKET(pnd->sd);
 526         PMIX_RELEASE(pnd);
 527         return;
 528     }
 529 
 530     /* get their gds module */
 531     PMIX_STRNLEN(msglen, ptr, len);
 532     if (msglen < len) {
 533         gds = ptr;
 534         ptr += strlen(gds) + 1;
 535         len -= strlen(gds) + 1;
 536     } else {
 537         free(msg);
 538         CLOSE_THE_SOCKET(pnd->sd);
 539         PMIX_RELEASE(pnd);
 540         return;
 541     }
 542 
 543     pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
 544                         "connect-ack recvd from peer %s:%d:%s on socket %d",
 545                         nspace, rank, version, pnd->sd);
 546 
 547     /* see if we know this nspace */
 548     nptr = NULL;
 549     PMIX_LIST_FOREACH(tmp, &pmix_server_globals.nspaces, pmix_namespace_t) {
 550         if (0 == strcmp(tmp->nspace, nspace)) {
 551             nptr = tmp;
 552             break;
 553         }
 554     }
 555     if (NULL == nptr) {
 556         /* we don't know this namespace, reject it */
 557         free(msg);
 558         /* send an error reply to the client */
 559         rc = PMIX_ERR_NOT_FOUND;
 560         goto error;
 561     }
 562 
 563     /* see if we have this peer in our list */
 564     info = NULL;
 565     found = false;
 566     PMIX_LIST_FOREACH(info, &nptr->ranks, pmix_rank_info_t) {
 567         if (info->pname.rank == rank) {
 568             found = true;
 569             break;
 570         }
 571     }
 572     if (!found) {
 573         /* rank unknown, reject it */
 574         free(msg);
 575         /* send an error reply to the client */
 576         rc = PMIX_ERR_NOT_FOUND;
 577         goto error;
 578     }
 579     /* a peer can connect on multiple sockets since it can fork/exec
 580      * a child that also calls PMIx_Init, so add it here if necessary.
 581      * Create the tracker for this peer */
 582     psave = PMIX_NEW(pmix_peer_t);
 583     if (NULL == psave) {
 584         free(msg);
 585         rc = PMIX_ERR_NOMEM;
 586         goto error;
 587     }
 588     /* mark it as being a client of the correct type */
 589     if (1 == major) {
 590         psave->proc_type = PMIX_PROC_CLIENT | PMIX_PROC_V1;
 591     } else if (2 == major && 0 == minor) {
 592         psave->proc_type = PMIX_PROC_CLIENT | PMIX_PROC_V20;
 593     } else if (2 == major && 1 == minor) {
 594         psave->proc_type = PMIX_PROC_CLIENT | PMIX_PROC_V21;
 595     } else if (3 == major) {
 596         psave->proc_type = PMIX_PROC_CLIENT | PMIX_PROC_V3;
 597     } else {
 598         /* we don't recognize this version */
 599         pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
 600                             "connection request from client of unrecognized version %s", version);
 601         free(msg);
 602         PMIX_RELEASE(psave);
 603         CLOSE_THE_SOCKET(pnd->sd);
 604         PMIX_RELEASE(pnd);
 605         return;
 606     }
 607     /* save the protocol */
 608     psave->protocol = pnd->protocol;
 609     /* add the nspace tracker */
 610     PMIX_RETAIN(nptr);
 611     psave->nptr = nptr;
 612     PMIX_RETAIN(info);
 613     psave->info = info;
 614     /* save the epilog info */
 615     psave->epilog.uid = info->uid;
 616     psave->epilog.gid = info->gid;
 617     nptr->epilog.uid = info->uid;
 618     nptr->epilog.gid = info->gid;
 619     info->proc_cnt++; /* increase number of processes on this rank */
 620     psave->sd = pnd->sd;
 621     if (0 > (psave->index = pmix_pointer_array_add(&pmix_server_globals.clients, psave))) {
 622         free(msg);
 623         info->proc_cnt--;
 624         PMIX_RELEASE(info);
 625         PMIX_RELEASE(psave);
 626         /* probably cannot send an error reply if we are out of memory */
 627         CLOSE_THE_SOCKET(pnd->sd);
 628         PMIX_RELEASE(pnd);
 629         return;
 630     }
 631     info->peerid = psave->index;
 632 
 633     /* get the appropriate compatibility modules */
 634     nptr->compat.psec = pmix_psec_base_assign_module(sec);
 635     if (NULL == nptr->compat.psec) {
 636         free(msg);
 637         info->proc_cnt--;
 638         PMIX_RELEASE(info);
 639         pmix_pointer_array_set_item(&pmix_server_globals.clients, psave->index, NULL);
 640         PMIX_RELEASE(psave);
 641         /* send an error reply to the client */
 642         goto error;
 643     }
 644 
 645     /* set the bfrops module to match this peer */
 646     nptr->compat.bfrops = pmix_bfrops_base_assign_module(bfrops);
 647     if (NULL == nptr->compat.bfrops) {
 648         free(msg);
 649         info->proc_cnt--;
 650         PMIX_RELEASE(info);
 651         pmix_pointer_array_set_item(&pmix_server_globals.clients, psave->index, NULL);
 652         PMIX_RELEASE(psave);
 653        /* send an error reply to the client */
 654         goto error;
 655     }
 656     /* set the buffer type */
 657     nptr->compat.type = bftype;
 658 
 659     /* set the gds module to match this peer */
 660     if (NULL != gds) {
 661         PMIX_INFO_LOAD(&ginfo, PMIX_GDS_MODULE, gds, PMIX_STRING);
 662         nptr->compat.gds = pmix_gds_base_assign_module(&ginfo, 1);
 663         PMIX_INFO_DESTRUCT(&ginfo);
 664     } else {
 665         nptr->compat.gds = pmix_gds_base_assign_module(NULL, 0);
 666     }
 667     if (NULL == nptr->compat.gds) {
 668         free(msg);
 669         info->proc_cnt--;
 670         pmix_pointer_array_set_item(&pmix_server_globals.clients, psave->index, NULL);
 671         PMIX_RELEASE(psave);
 672         /* send an error reply to the client */
 673         goto error;
 674     }
 675 
 676     /* if we haven't previously stored the version for this
 677      * nspace, do so now */
 678     if (!nptr->version_stored) {
 679         PMIX_INFO_LOAD(&ginfo, PMIX_BFROPS_MODULE, nptr->compat.bfrops->version, PMIX_STRING);
 680         PMIX_GDS_CACHE_JOB_INFO(rc, pmix_globals.mypeer, nptr, &ginfo, 1);
 681         PMIX_INFO_DESTRUCT(&ginfo);
 682         nptr->version_stored = true;
 683     }
 684 
 685     /* the choice of PTL module was obviously made by the connecting
 686      * tool as we received this request via that channel, so simply
 687      * record it here for future use */
 688     nptr->compat.ptl = &pmix_ptl_usock_module;
 689 
 690     /* validate the connection - the macro will send the status result to the client */
 691     PMIX_PSEC_VALIDATE_CONNECTION(rc, psave, NULL, 0, NULL, 0, &cred);
 692     /* now done with the msg */
 693     free(msg);
 694 
 695     if (PMIX_SUCCESS != rc) {
 696         pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
 697                             "validation of client credentials failed: %s",
 698                             PMIx_Error_string(rc));
 699         info->proc_cnt--;
 700         PMIX_RELEASE(info);
 701         pmix_pointer_array_set_item(&pmix_server_globals.clients, psave->index, NULL);
 702         PMIX_RELEASE(psave);
 703         /* error reply was sent by the above macro */
 704         CLOSE_THE_SOCKET(pnd->sd);
 705         PMIX_RELEASE(pnd);
 706         return;
 707     }
 708 
 709     /* send the client's array index */
 710     if (PMIX_SUCCESS != (rc = pmix_ptl_base_send_blocking(pnd->sd, (char*)&psave->index, sizeof(int)))) {
 711         PMIX_ERROR_LOG(rc);
 712         info->proc_cnt--;
 713         PMIX_RELEASE(info);
 714         pmix_pointer_array_set_item(&pmix_server_globals.clients, psave->index, NULL);
 715         PMIX_RELEASE(psave);
 716         CLOSE_THE_SOCKET(pnd->sd);
 717         PMIX_RELEASE(pnd);
 718         return;
 719     }
 720 
 721     pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
 722                         "connect-ack from client completed");
 723 
 724     /* let the host server know that this client has connected */
 725     if (NULL != pmix_host_server.client_connected) {
 726         pmix_strncpy(proc.nspace, psave->info->pname.nspace, PMIX_MAX_NSLEN);
 727         proc.rank = psave->info->pname.rank;
 728         rc = pmix_host_server.client_connected(&proc, psave->info->server_object, NULL, NULL);
 729         if (PMIX_SUCCESS != rc && PMIX_OPERATION_SUCCEEDED != rc) {
 730             PMIX_ERROR_LOG(rc);
 731             info->proc_cnt--;
 732             PMIX_RELEASE(info);
 733             pmix_pointer_array_set_item(&pmix_server_globals.clients, psave->index, NULL);
 734             PMIX_RELEASE(psave);
 735             CLOSE_THE_SOCKET(pnd->sd);
 736         }
 737     }
 738 
 739     /* start the events for this client */
 740     pmix_event_assign(&psave->recv_event, pmix_globals.evbase, pnd->sd,
 741                       EV_READ|EV_PERSIST, pmix_usock_recv_handler, psave);
 742     pmix_event_add(&psave->recv_event, NULL);
 743     psave->recv_ev_active = true;
 744     pmix_event_assign(&psave->send_event, pmix_globals.evbase, pnd->sd,
 745                       EV_WRITE|EV_PERSIST, pmix_usock_send_handler, psave);
 746     pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
 747                         "pmix:server client %s:%u has connected on socket %d",
 748                         psave->info->pname.nspace, psave->info->pname.rank, psave->sd);
 749 
 750     PMIX_RELEASE(pnd);
 751     return;
 752 
 753   error:
 754     /* send an error reply to the client */
 755     if (PMIX_SUCCESS != pmix_ptl_base_send_blocking(pnd->sd, (char*)&rc, sizeof(int))) {
 756         PMIX_ERROR_LOG(rc);
 757     }
 758     CLOSE_THE_SOCKET(pnd->sd);
 759     PMIX_RELEASE(pnd);
 760     return;
 761 }
/* [<][>][^][v][top][bottom][index][help] */
root/opal/mca/pmix/pmix4x/pmix/src/mca/ptl/usock/ptl_usock_component.c

DEFINITIONS
