Provide callback function for client_connected2

rhc54 · rhc54 · commit 92f31dd90fbb · 2025-05-10T09:11:48.000-06:00
The client_connected2 upcall provides a callback function, but the current code passed a NULL for it, thereby disabling that feature. Implement and provide a callback function for it. Signed-off-by: Ralph Castain <rhc@pmix.org> (cherry picked from commit 611567a)
diff --git a/examples/client2.c b/examples/client2.c
@@ -15,7 +15,7 @@
  * Copyright (c) 2011      Oak Ridge National Labs.  All rights reserved.
  * Copyright (c) 2013-2020 Intel, Inc.  All rights reserved.
  * Copyright (c) 2015      Mellanox Technologies, Inc.  All rights reserved.
- * Copyright (c) 2021      Nanook Consulting.  All rights reserved.
+ * Copyright (c) 2021-2025 Nanook Consulting  All rights reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -76,9 +76,10 @@ int main(int argc, char **argv)
 {
     pmix_status_t rc;
     pmix_value_t value;
-    pmix_value_t *val, *vptr;
+    pmix_value_t *val;
     pmix_proc_t proc;
     uint32_t nprocs, n, k;
+    uint64_t *u64;
     pmix_info_t *info;
     bool flag;
     mylock_t mylock;
@@ -130,14 +131,13 @@ int main(int argc, char **argv)
     fprintf(stderr, "Client %s:%d job size %d\n", myproc.nspace, myproc.rank, nprocs);
 
     /* put a data array of pmix_value's */
-    val = (pmix_value_t *) malloc(32 * sizeof(pmix_value_t));
+    u64 = (uint64_t *) malloc(32 * sizeof(uint64_t));
     for (n = 0; n < 32; n++) {
-        val[n].type = PMIX_UINT64;
-        val[n].data.uint64 = 2 * n;
+        u64[n] = 2 * n;
     }
-    da.type = PMIX_VALUE;
+    da.type = PMIX_UINT64;
     da.size = 32;
-    da.array = val;
+    da.array = u64;
     value.type = PMIX_DATA_ARRAY;
     value.data.darray = &da;
     rc = PMIx_Put(PMIX_GLOBAL, "test-key", &value);
@@ -189,7 +189,7 @@ int main(int argc, char **argv)
             PMIX_VALUE_RELEASE(val);
             goto done;
         }
-        if (PMIX_VALUE != dptr->type) {
+        if (PMIX_UINT64 != dptr->type) {
             fprintf(stderr,
                     "Client ns %s rank %d: PMIx_Get %d returned wrong array value type %d\n",
                     myproc.nspace, myproc.rank, proc.rank, dptr->type);
@@ -203,17 +203,12 @@ int main(int argc, char **argv)
             PMIX_VALUE_RELEASE(val);
             goto done;
         }
-        vptr = (pmix_value_t *) dptr->array;
+        u64 = (uint64_t *) dptr->array;
         for (k = 0; k < 32; k++) {
-            if (PMIX_UINT64 != vptr[k].type) {
-                fprintf(stderr, "Client ns %s rank %d: PMIx_Get %d returned wrong type: %d\n",
-                        myproc.nspace, myproc.rank, proc.rank, vptr[k].type);
-                PMIX_VALUE_RELEASE(val);
-                goto done;
-            }
-            if (2 * k != vptr[k].data.uint64) {
-                fprintf(stderr, "Client ns %s rank %d: PMIx_Get %d returned wrong value: %lu\n",
-                        myproc.nspace, myproc.rank, proc.rank, (unsigned long) vptr[k].data.uint64);
+            if ((2 * k) != u64[k]) {
+            fprintf(stderr,
+                    "Client ns %s rank %d: PMIx_Get %d returned wrong value: %lu\n",
+                    myproc.nspace, myproc.rank, proc.rank, (unsigned long) u64[k]);
                 PMIX_VALUE_RELEASE(val);
                 goto done;
             }
diff --git a/src/mca/psec/psec.h b/src/mca/psec/psec.h
@@ -6,7 +6,7 @@
  * Copyright (c) 2015      Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  * Copyright (c) 2019      Mellanox Technologies, Inc. All rights reserved.
- * Copyright (c) 2021-2022 Nanook Consulting.  All rights reserved.
+ * Copyright (c) 2021-2025 Nanook Consulting  All rights reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -164,7 +164,7 @@ PMIX_EXPORT pmix_psec_module_t *pmix_psec_base_assign_module(const char *options
         }                                                                                          \
     } while (0)
 
-#define PMIX_PSEC_SERVER_HANDSHAKE_IFNEED(r, p, d, nd, in, nin, c)                    \
+#define PMIX_PSEC_SERVER_HANDSHAKE_IFNEED(r, p)                                       \
     if (PMIX_ERR_READY_FOR_HANDSHAKE == r) {                                          \
         int _r;                                                                       \
         /* execute the handshake if the security mode calls for it */                 \
diff --git a/src/mca/ptl/base/ptl_base_connection_hdlr.c b/src/mca/ptl/base/ptl_base_connection_hdlr.c
@@ -51,14 +51,113 @@ static void cnct_cbfunc(pmix_status_t status, pmix_proc_t *proc, void *cbdata);
 static void _check_cached_events(pmix_peer_t *peer);
 static pmix_status_t process_tool_request(pmix_pending_connection_t *pnd, char *mg, size_t cnt);
 
+// Local objects
+typedef struct {
+    pmix_object_t super;
+    pmix_event_t ev;
+    pmix_status_t status;
+    pmix_status_t reply;
+    pmix_pending_connection_t *pnd;
+    char *blob;
+    pmix_peer_t *peer;
+} cnct_hdlr_t;
+static void chcon(cnct_hdlr_t *p)
+{
+    memset(&p->ev, 0, sizeof(pmix_event_t));
+    p->pnd = NULL;
+    p->blob = NULL;
+    p->peer = NULL;
+}
+static void chdes(cnct_hdlr_t *p)
+{
+    if (NULL != p->pnd) {
+        PMIX_RELEASE(p->pnd);
+    }
+    if (NULL != p->blob) {
+        free(p->blob);
+    }
+}
+static PMIX_CLASS_INSTANCE(cnct_hdlr_t,
+                           pmix_object_t,
+                           chcon, chdes);
+
+static void _cnct_complete(int sd, short args, void *cbdata)
+{
+    cnct_hdlr_t *ch = (cnct_hdlr_t *) cbdata;
+    uint32_t u32;
+    pmix_status_t rc;
+    PMIX_HIDE_UNUSED_PARAMS(sd, args);
+
+    /* tell the client all is good */
+    u32 = htonl(ch->reply);
+    rc = pmix_ptl_base_send_blocking(ch->pnd->sd, (char *) &u32, sizeof(uint32_t));
+    if (PMIX_SUCCESS != rc) {
+        PMIX_ERROR_LOG(rc);
+        goto error;
+    }
+    /* If needed, perform the handshake. The macro will update reply */
+    PMIX_PSEC_SERVER_HANDSHAKE_IFNEED(ch->reply, ch->peer);
+
+    /* It is possible that connection validation failed */
+    if (PMIX_SUCCESS != ch->reply) {
+        pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
+                            "validation of client connection failed");
+        goto error;
+    }
+
+    /* send the client's array index */
+    u32 = htonl(ch->peer->index);
+    rc = pmix_ptl_base_send_blocking(ch->pnd->sd, (char *) &u32, sizeof(uint32_t));
+    if (PMIX_SUCCESS != rc) {
+        PMIX_ERROR_LOG(rc);
+        goto error;
+    }
+
+    pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
+                        "connect-ack from client completed");
+
+    pmix_ptl_base_set_nonblocking(ch->pnd->sd);
+
+    /* start the events for this client */
+    pmix_event_assign(&ch->peer->recv_event, pmix_globals.evbase, ch->pnd->sd, EV_READ | EV_PERSIST,
+                      pmix_ptl_base_recv_handler, ch->peer);
+    pmix_event_add(&ch->peer->recv_event, NULL);
+    ch->peer->recv_ev_active = true;
+    pmix_event_assign(&ch->peer->send_event, pmix_globals.evbase, ch->pnd->sd, EV_WRITE | EV_PERSIST,
+                      pmix_ptl_base_send_handler, ch->peer);
+    pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
+                        "pmix:server client %s:%u has connected on socket %d",
+                        ch->peer->info->pname.nspace, ch->peer->info->pname.rank, ch->peer->sd);
+
+    /* check the cached events and update the client */
+    _check_cached_events(ch->peer);
+    PMIX_RELEASE(ch);
+    return;
+
+error:
+    if (NULL != ch->peer) {
+        pmix_pointer_array_set_item(&pmix_server_globals.clients, ch->peer->index, NULL);
+        PMIX_RELEASE(ch->peer);
+    }
+    CLOSE_THE_SOCKET(ch->pnd->sd);
+    PMIX_RELEASE(ch);
+}
+
+static void _connect_complete(pmix_status_t status, void *cbdata)
+{
+    cnct_hdlr_t *ch = (cnct_hdlr_t *) cbdata;
+    /* need to thread-shift this response */
+    ch->status = status;
+    PMIX_THREADSHIFT(ch, _cnct_complete);
+}
+
 void pmix_ptl_base_connection_handler(int sd, short args, void *cbdata)
 {
     pmix_pending_connection_t *pnd = (pmix_pending_connection_t *) cbdata;
     pmix_ptl_hdr_t hdr;
     pmix_peer_t *peer = NULL;
     pmix_status_t rc, reply;
     char *msg = NULL, *mg, *p, *blob = NULL;
-    uint32_t u32;
     size_t cnt;
     size_t len = 0;
     pmix_namespace_t *nptr, *tmp;
@@ -67,6 +166,7 @@ void pmix_ptl_base_connection_handler(int sd, short args, void *cbdata)
     pmix_info_t ginfo;
     pmix_byte_object_t cred;
     uint8_t major, minor, release;
+    cnct_hdlr_t *ch;
 
     /* acquire the object */
     PMIX_ACQUIRE_OBJECT(pnd);
@@ -370,72 +470,47 @@ void pmix_ptl_base_connection_handler(int sd, short args, void *cbdata)
         goto error;
     }
 
-    pmix_output_verbose(2, pmix_ptl_base_framework.framework_output, "client connection validated");
+    pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
+                        "client connection validated");
 
-    /* tell the client all is good */
-    u32 = htonl(reply);
-    if (PMIX_SUCCESS
-        != (rc = pmix_ptl_base_send_blocking(pnd->sd, (char *) &u32, sizeof(uint32_t)))) {
-        PMIX_ERROR_LOG(rc);
-        goto error;
-    }
-    /* If needed, perform the handshake. The macro will update reply */
-    PMIX_PSEC_SERVER_HANDSHAKE_IFNEED(reply, peer, NULL, 0, NULL, NULL, &cred);
 
-    /* It is possible that connection validation failed */
-    if (PMIX_SUCCESS != reply) {
-        pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
-                            "validation of client connection failed");
-        goto error;
-    }
-
-    /* send the client's array index */
-    u32 = htonl(peer->index);
-    if (PMIX_SUCCESS
-        != (rc = pmix_ptl_base_send_blocking(pnd->sd, (char *) &u32, sizeof(uint32_t)))) {
-        PMIX_ERROR_LOG(rc);
-        goto error;
-    }
-
-    pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
-                        "connect-ack from client completed");
+    // prep for processing
+    ch = PMIX_NEW(cnct_hdlr_t);
+    ch->peer = peer;
+    ch->pnd = pnd;
+    ch->reply = reply;
+    ch->blob = blob;
 
     /* let the host server know that this client has connected */
     if (NULL != pmix_host_server.client_connected2) {
         PMIX_LOAD_PROCID(&proc, peer->info->pname.nspace, peer->info->pname.rank);
-        rc = pmix_host_server.client_connected2(&proc, peer->info->server_object, NULL, 0, NULL,
-                                                NULL);
-        if (PMIX_SUCCESS != rc && PMIX_OPERATION_SUCCEEDED != rc) {
+        rc = pmix_host_server.client_connected2(&proc, peer->info->server_object, NULL, 0,
+                                                _connect_complete, ch);
+        if (PMIX_OPERATION_SUCCEEDED == rc) {
+            ch->status = PMIX_SUCCESS;
+            _cnct_complete(0, 0, ch);
+            return;
+        }
+        if (PMIX_SUCCESS != rc) {
             PMIX_ERROR_LOG(rc);
+            goto error;
         }
     } else if (NULL != pmix_host_server.client_connected) {
         PMIX_LOAD_PROCID(&proc, peer->info->pname.nspace, peer->info->pname.rank);
-        rc = pmix_host_server.client_connected(&proc, peer->info->server_object, NULL, NULL);
-        if (PMIX_SUCCESS != rc && PMIX_OPERATION_SUCCEEDED != rc) {
+        rc = pmix_host_server.client_connected(&proc, peer->info->server_object, _connect_complete, ch);
+        if (PMIX_OPERATION_SUCCEEDED == rc) {
+            ch->status = PMIX_SUCCESS;
+            _cnct_complete(0, 0, ch);
+            return;
+        }
+        if (PMIX_SUCCESS != rc) {
             PMIX_ERROR_LOG(rc);
             goto error;
         }
-    }
-
-    pmix_ptl_base_set_nonblocking(pnd->sd);
-
-    /* start the events for this client */
-    pmix_event_assign(&peer->recv_event, pmix_globals.evbase, pnd->sd, EV_READ | EV_PERSIST,
-                      pmix_ptl_base_recv_handler, peer);
-    pmix_event_add(&peer->recv_event, NULL);
-    peer->recv_ev_active = true;
-    pmix_event_assign(&peer->send_event, pmix_globals.evbase, pnd->sd, EV_WRITE | EV_PERSIST,
-                      pmix_ptl_base_send_handler, peer);
-    pmix_output_verbose(2, pmix_ptl_base_framework.framework_output,
-                        "pmix:server client %s:%u has connected on socket %d",
-                        peer->info->pname.nspace, peer->info->pname.rank, peer->sd);
-    PMIX_RELEASE(pnd);
-
-    /* check the cached events and update the client */
-    _check_cached_events(peer);
-    if (NULL != blob) {
-        free(blob);
-        blob = NULL;
+    } else {
+        // if neither of those conditions are met, then we simply assume the host is ready
+        ch->status = PMIX_SUCCESS;
+        _cnct_complete(0, 0, ch);
     }
 
     return;
@@ -448,9 +523,6 @@ void pmix_ptl_base_connection_handler(int sd, short args, void *cbdata)
     if (NULL != msg) {
         free(msg);
     }
-    if (NULL != blob) {
-        free(blob);
-    }
     if (NULL != peer) {
         pmix_pointer_array_set_item(&pmix_server_globals.clients, peer->index, NULL);
         PMIX_RELEASE(peer);
@@ -610,7 +682,7 @@ static void process_cbfunc(int sd, short args, void *cbdata)
     }
 
     /* If needed perform the handshake. The macro will update reply */
-    PMIX_PSEC_SERVER_HANDSHAKE_IFNEED(reply, peer, NULL, 0, NULL, NULL, &cred);
+    PMIX_PSEC_SERVER_HANDSHAKE_IFNEED(reply, peer);
 
     /* If verification wasn't successful - stop here */
     if (PMIX_SUCCESS != reply) {
