1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361
|
/*
* Copyright (c) 2004-2010 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2011 The University of Tennessee and The University
* of Tennessee Research Foundation. All rights
* reserved.
* Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
* University of Stuttgart. All rights reserved.
* Copyright (c) 2004-2005 The Regents of the University of California.
* All rights reserved.
* Copyright (c) 2006-2013 Los Alamos National Security, LLC.
* All rights reserved.
* Copyright (c) 2009-2012 Cisco Systems, Inc. All rights reserved.
* 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) 2019 IBM Corporation. All rights reserved.
* Copyright (c) 2021-2025 Nanook Consulting All rights reserved.
* $COPYRIGHT$
*
* Additional copyrights may follow
*
* $HEADER$
*
*/
#define _GNU_SOURCE
#include <pthread.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <unistd.h>
#include <libgen.h>
#include <pmix.h>
#include "examples.h"
static pmix_proc_t myproc;
static void notification_fn(size_t evhdlr_registration_id, pmix_status_t status,
const pmix_proc_t *source, pmix_info_t info[], size_t ninfo,
pmix_info_t results[], size_t nresults,
pmix_event_notification_cbfunc_fn_t cbfunc, void *cbdata)
{
EXAMPLES_HIDE_UNUSED_PARAMS(evhdlr_registration_id, source,
info, ninfo, results, nresults,
cbfunc, cbdata);
fprintf(stderr, "Client %s:%d NOTIFIED with status %d\n", myproc.nspace, myproc.rank, status);
}
static void op_callbk(pmix_status_t status, void *cbdata)
{
mylock_t *lock = (mylock_t *) cbdata;
lock->status = status;
DEBUG_WAKEUP_THREAD(lock);
}
static void errhandler_reg_callbk(pmix_status_t status, size_t errhandler_ref, void *cbdata)
{
mylock_t *lock = (mylock_t *) cbdata;
EXAMPLES_HIDE_UNUSED_PARAMS(errhandler_ref);
lock->status = status;
DEBUG_WAKEUP_THREAD(lock);
}
int main(int argc, char **argv)
{
int rc;
pmix_value_t *val = NULL;
uint32_t nprocs;
pmix_proc_t proc, *parray;
pmix_value_t value;
mylock_t lock;
pmix_info_t *results = NULL, info[3], tinfo;
size_t nresults, cid, n, m, psize=0, maxprocs = 2;
pmix_data_array_t dry;
char hostname[1024], dir[1024];
char tmp[1024];
pmix_app_t app;
char nsp2[PMIX_MAX_NSLEN + 1];
EXAMPLES_HIDE_UNUSED_PARAMS(argc, argv);
gethostname(hostname, 1024);
if (NULL == getcwd(dir, 1024)) {
fprintf(stderr, "Getcwd failure\n");
exit(1);
}
if (1 < argc) {
maxprocs = atoi(argv[1]);
}
/* init us */
if (PMIX_SUCCESS != (rc = PMIx_Init(&myproc, NULL, 0))) {
fprintf(stderr, "Client ns %s rank %d: PMIx_Init failed: %s\n", myproc.nspace, myproc.rank,
PMIx_Error_string(rc));
exit(0);
}
fprintf(stderr, "Client ns %s rank %d: Running on host %s\n", myproc.nspace, myproc.rank, hostname);
PMIX_PROC_CONSTRUCT(&proc);
PMIX_LOAD_PROCID(&proc, myproc.nspace, PMIX_RANK_WILDCARD);
/* get our job size */
if (PMIX_SUCCESS != (rc = PMIx_Get(&proc, PMIX_JOB_SIZE, NULL, 0, &val))) {
fprintf(stderr, "Client ns %s rank %d: PMIx_Get job size failed: %s\n", myproc.nspace,
myproc.rank, PMIx_Error_string(rc));
goto done;
}
nprocs = val->data.uint32;
PMIX_VALUE_RELEASE(val);
if (nprocs < 2) {
if (0 == myproc.rank) {
fprintf(stderr, "This example requires a minimum of 2 processes\n");
}
goto done;
}
fprintf(stderr, "Client %s:%d job size %u\n", myproc.nspace, myproc.rank, nprocs);
/* register our default errhandler */
DEBUG_CONSTRUCT_LOCK(&lock);
PMIx_Register_event_handler(NULL, 0, NULL, 0, notification_fn, errhandler_reg_callbk,
(void *) &lock);
DEBUG_WAIT_THREAD(&lock);
rc = lock.status;
DEBUG_DESTRUCT_LOCK(&lock);
if (PMIX_SUCCESS != rc) {
goto done;
}
// see if we are the parent job
if (NULL == getenv("CHILD")) {
// put some modex-like data
(void) snprintf(tmp, 1024, "%s-%u-remote", myproc.nspace, myproc.rank);
value.type = PMIX_UINT64;
value.data.uint64 = 1234UL + (unsigned long) myproc.rank;
if (PMIX_SUCCESS != (rc = PMIx_Put(PMIX_GLOBAL, tmp, &value))) {
fprintf(stderr, "Client ns %s rank %d: PMIx_Put failed: %d\n", myproc.nspace,
myproc.rank, rc);
goto done;
}
// commit it
rc = PMIx_Commit();
if (PMIX_SUCCESS != rc) {
fprintf(stderr, "Client ns %s rank %d: PMIx_Commit failed: %s\n",
myproc.nspace, myproc.rank, PMIx_Error_string(rc));
goto done;
}
// do our own exchange
PMIX_PROC_CONSTRUCT(&proc);
PMIX_LOAD_PROCID(&proc, myproc.nspace, PMIX_RANK_WILDCARD);
if (PMIX_SUCCESS != (rc = PMIx_Fence(&proc, 1, NULL, 0))) {
fprintf(stderr, "Client ns %s rank %d: PMIx_Fence failed: %s\n",
myproc.nspace, myproc.rank, PMIx_Error_string(rc));
goto done;
}
// parent rank=0 will call spawn
if (0 == myproc.rank) {
PMIX_APP_CONSTRUCT(&app);
app.cmd = strdup(argv[0]);
app.maxprocs = maxprocs;
app.argv = (char **) malloc(2 * sizeof(char *));
app.argv[0] = strdup(basename(argv[0]));
app.argv[1] = NULL;
app.env = (char **) malloc(2 * sizeof(char *));
app.env[0] = strdup("CHILD=3");
app.env[1] = NULL;
fprintf(stderr, "Client ns %s rank %d: calling PMIx_Spawn\n", myproc.nspace, myproc.rank);
PMIX_INFO_LOAD(&info[0], PMIX_MAPBY, "node", PMIX_STRING);
if (PMIX_SUCCESS != (rc = PMIx_Spawn(info, 1, &app, 1, nsp2))) {
fprintf(stderr, "Client ns %s rank %d: PMIx_Spawn failed: %d\n", myproc.nspace,
myproc.rank, rc);
goto done;
}
PMIX_APP_DESTRUCT(&app);
// now bootstrap the group
// ask for a new context ID
PMIX_INFO_LOAD(&info[0], PMIX_GROUP_ASSIGN_CONTEXT_ID, NULL, PMIX_BOOL);
// two procs are performing the bootstrap
n = 2;
PMIX_INFO_LOAD(&info[1], PMIX_GROUP_BOOTSTRAP, &n, PMIX_SIZE);
// add the rest of the parent job
PMIX_DATA_ARRAY_CONSTRUCT(&dry, 1, PMIX_PROC);
parray = (pmix_proc_t*)dry.array;
PMIX_LOAD_PROCID(&parray[0], myproc.nspace, PMIX_RANK_WILDCARD);
PMIX_INFO_LOAD(&info[2], PMIX_GROUP_ADD_MEMBERS, &dry, PMIX_DATA_ARRAY);
PMIX_DATA_ARRAY_DESTRUCT(&dry);
// construct the group
rc = PMIx_Group_construct("ourgroup", &myproc, 1, info, 3, &results, &nresults);
if (PMIX_SUCCESS != rc) {
fprintf(stderr, "Client ns %s rank %d: PMIx_Group_construct failed: %s\n",
myproc.nspace, myproc.rank, PMIx_Error_string(rc));
goto done;
}
} else {
// we will participate in the group assembly with
// the spawned children
rc = PMIx_Group_construct("ourgroup", NULL, 0, NULL, 0, &results, &nresults);
if (PMIX_SUCCESS != rc) {
fprintf(stderr, "Client ns %s rank %d: PMIx_Group_construct failed: %s\n",
myproc.nspace, myproc.rank, PMIx_Error_string(rc));
goto done;
}
// process results to get final membership
}
} else {
// we are the child job - put some data
(void) snprintf(tmp, 1024, "%s-%u-remote", myproc.nspace, myproc.rank);
value.type = PMIX_UINT64;
value.data.uint64 = 1234UL + (unsigned long) myproc.rank;
if (PMIX_SUCCESS != (rc = PMIx_Put(PMIX_GLOBAL, tmp, &value))) {
fprintf(stderr, "Client ns %s rank %d: PMIx_Put failed: %d\n", myproc.nspace,
myproc.rank, rc);
goto done;
}
// commit it
rc = PMIx_Commit();
if (PMIX_SUCCESS != rc) {
fprintf(stderr, "Client ns %s rank %d: PMIx_Commit failed: %s\n",
myproc.nspace, myproc.rank, PMIx_Error_string(rc));
goto done;
}
// do our own exchange
PMIX_PROC_CONSTRUCT(&proc);
PMIX_LOAD_PROCID(&proc, myproc.nspace, PMIX_RANK_WILDCARD);
if (PMIX_SUCCESS != (rc = PMIx_Fence(&proc, 1, NULL, 0))) {
fprintf(stderr, "Client ns %s rank %d: PMIx_Fence failed: %s\n",
myproc.nspace, myproc.rank, PMIx_Error_string(rc));
goto done;
}
// rank=0 will setup the group assembly with the parent
if (0 == myproc.rank) {
// two procs are performing the bootstrap
n = 2;
PMIX_INFO_LOAD(&info[0], PMIX_GROUP_BOOTSTRAP, &n, PMIX_SIZE);
// add the rest of the child job
PMIX_DATA_ARRAY_CONSTRUCT(&dry, 1, PMIX_PROC);
parray = (pmix_proc_t*)dry.array;
PMIX_LOAD_PROCID(&parray[0], myproc.nspace, PMIX_RANK_WILDCARD);
PMIX_INFO_LOAD(&info[1], PMIX_GROUP_ADD_MEMBERS, &dry, PMIX_DATA_ARRAY);
PMIX_DATA_ARRAY_DESTRUCT(&dry);
// construct the group
rc = PMIx_Group_construct("ourgroup", &myproc, 1, info, 2, &results, &nresults);
if (PMIX_SUCCESS != rc) {
fprintf(stderr, "Client ns %s rank %d: PMIx_Group_construct failed: %s\n",
myproc.nspace, myproc.rank, PMIx_Error_string(rc));
goto done;
}
} else {
// wait for group assembly to complete
rc = PMIx_Group_construct("ourgroup", NULL, 0, NULL, 0, &results, &nresults);
if (PMIX_SUCCESS != rc) {
fprintf(stderr, "Client ns %s rank %d: PMIx_Group_construct failed: %s\n",
myproc.nspace, myproc.rank, PMIx_Error_string(rc));
goto done;
}
}
}
/* we should have a single results object */
if (NULL != results) {
cid = 0;
parray = NULL;
for (n=0; n < nresults; n++) {
if (PMIX_CHECK_KEY(&results[n], PMIX_GROUP_CONTEXT_ID)) {
rc = PMIx_Value_get_number(&results[n].value, &cid, PMIX_SIZE);
fprintf(stderr, "%s:%d Group construct complete with status %s KEY %s CID %lu\n",
myproc.nspace, myproc.rank, PMIx_Error_string(rc), results[n].key, (unsigned long) cid);
} else if (PMIX_CHECK_KEY(&results[n], PMIX_GROUP_MEMBERSHIP)) {
parray = (pmix_proc_t*)results[n].value.data.darray->array;
psize = results[n].value.data.darray->size;
fprintf(stderr, "[%s:%u] NUM MEMBERS: %u MEMBERSHIP:\n", myproc.nspace, myproc.rank, (unsigned)psize);
for (m=0; m < psize; m++) {
fprintf(stderr, "\t%s:%u\n", parray[m].nspace, parray[m].rank);
}
}
}
if (NULL == parray) {
fprintf(stderr, "%s:%d Membership not returned\n", myproc.nspace, myproc.rank);
goto done;
}
} else {
fprintf(stderr, "%s:%d No returned results\n", myproc.nspace, myproc.rank);
goto done;
}
// check for modex data to have been exchanged
PMIX_INFO_CONSTRUCT(&tinfo);
PMIX_INFO_LOAD(&tinfo, PMIX_IMMEDIATE, NULL, PMIX_BOOL);
for (n=0; n < psize; n++) {
// ignore my own nspace
if (PMIX_CHECK_NSPACE(parray[n].nspace, myproc.nspace)) {
continue;
}
if (PMIX_RANK_WILDCARD == parray[n].rank) {
// get the size of the other nspace
PMIX_LOAD_PROCID(&proc, parray[n].nspace, PMIX_RANK_WILDCARD);
rc = PMIx_Get(&proc, PMIX_JOB_SIZE, &tinfo, 1, &val);
if (PMIX_SUCCESS != rc) {
fprintf(stderr, "Client ns %s rank %d: PMIx_Get job size failed for nspace %s: %s\n",
myproc.nspace, myproc.rank, parray[n].nspace, PMIx_Error_string(rc));
goto done;
}
nprocs = val->data.uint32;
PMIX_VALUE_RELEASE(val);
// cycle across the procs and get their modex data
for (m=0; m < nprocs; m++) {
proc.rank = m;
(void)snprintf(tmp, 1024, "%s-%u-remote", proc.nspace, proc.rank);
if (PMIX_SUCCESS != (rc = PMIx_Get(&proc, tmp, &tinfo, 1, &val))) {
fprintf(stderr, "Client ns %s rank %d: PMIx_Get %s failed: %s\n",
myproc.nspace, (int)n, tmp, PMIx_Error_string(rc));
} else {
fprintf(stderr, "Client ns %s rank %d: Get modex for proc %s:%u returned %lu\n",
myproc.nspace, myproc.rank, proc.nspace, proc.rank, (unsigned long)val->data.uint64);
}
}
}
}
done:
/* finalize us */
DEBUG_CONSTRUCT_LOCK(&lock);
PMIx_Deregister_event_handler(1, op_callbk, &lock);
DEBUG_WAIT_THREAD(&lock);
DEBUG_DESTRUCT_LOCK(&lock);
fprintf(stderr, "Client ns %s rank %d: Finalizing\n", myproc.nspace, myproc.rank);
if (PMIX_SUCCESS != (rc = PMIx_Finalize(NULL, 0))) {
fprintf(stderr, "Client ns %s rank %d:PMIx_Finalize failed: %s\n", myproc.nspace,
myproc.rank, PMIx_Error_string(rc));
} else {
fprintf(stderr, "Client ns %s rank %d:PMIx_Finalize successfully completed\n",
myproc.nspace, myproc.rank);
}
fprintf(stderr, "%s:%d COMPLETE\n", myproc.nspace, myproc.rank);
fflush(stderr);
return (0);
}
|