opal/mca/pmix/ext2x/ext2x_server

/* [<][>][^][v][top][bottom][index][help] */
This source file includes following definitions.
opal_opcbfunc
server_client_connected_fn
server_client_finalized_fn
server_abort_fn
_data_release
opmdx_response
server_fencenb_fn
server_dmodex_req_fn
server_publish_fn
opal_lkupcbfunc
server_lookup_fn
server_unpublish_fn
opal_spncbfunc
server_spawn_fn
server_connect_fn
server_disconnect_fn
server_register_events
server_deregister_events
server_notify_event
_info_rel
info_cbfunc
server_query
toolcbfunc
server_tool_connection
server_log
server_allocate
server_job_control
   1 /* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
   2 /*
   3  * Copyright (c) 2014-2017 Intel, Inc.  All rights reserved.
   4  * Copyright (c) 2014-2017 Research Organization for Information Science
   5  *                         and Technology (RIST). All rights reserved.
   6  * Copyright (c) 2014-2015 Mellanox Technologies, Inc.
   7  *                         All rights reserved.
   8  * Copyright (c) 2016 Cisco Systems, Inc.  All rights reserved.
   9  * $COPYRIGHT$
  10  *
  11  * Additional copyrights may follow
  12  *
  13  * $HEADER$
  14  */
  15 
  16 #include "opal_config.h"
  17 #include "opal/constants.h"
  18 #include "opal/types.h"
  19 
  20 #ifdef HAVE_STRING_H
  21 #include <string.h>
  22 #endif
  23 #ifdef HAVE_UNISTD_H
  24 #include <unistd.h>
  25 #endif
  26 
  27 #include "opal/dss/dss.h"
  28 #include "opal/mca/event/event.h"
  29 #include "opal/mca/hwloc/base/base.h"
  30 #include "opal/runtime/opal.h"
  31 #include "opal/runtime/opal_progress_threads.h"
  32 #include "opal/threads/threads.h"
  33 #include "opal/util/argv.h"
  34 #include "opal/util/error.h"
  35 #include "opal/util/output.h"
  36 #include "opal/util/proc.h"
  37 #include "opal/util/show_help.h"
  38 #include "opal/util/string_copy.h"
  39 #include "opal/mca/pmix/base/base.h"
  40 #include "ext2x.h"
  41 
  42 #include "pmix.h"
  43 #include "pmix_server.h"
  44 
  45 /****    N.O.R.T.H.B.O.U.N.D   I.N.T.E.R.F.A.C.E.S     ****/
  46 
  47 /* These are the interfaces used by the embedded PMIx server
  48  * to call up into ORTE for service requests */
  49 
  50 static pmix_status_t server_client_connected_fn(const pmix_proc_t *proc, void* server_object,
  51                                                 pmix_op_cbfunc_t cbfunc, void *cbdata);
  52 static pmix_status_t server_client_finalized_fn(const pmix_proc_t *proc, void* server_object,
  53                                                 pmix_op_cbfunc_t cbfunc, void *cbdata);
  54 static pmix_status_t server_abort_fn(const pmix_proc_t *proc, void *server_object,
  55                                      int status, const char msg[],
  56                                      pmix_proc_t procs[], size_t nprocs,
  57                                      pmix_op_cbfunc_t cbfunc, void *cbdata);
  58 static pmix_status_t server_fencenb_fn(const pmix_proc_t procs[], size_t nprocs,
  59                                        const pmix_info_t info[], size_t ninfo,
  60                                        char *data, size_t ndata,
  61                                        pmix_modex_cbfunc_t cbfunc, void *cbdata);
  62 static pmix_status_t server_dmodex_req_fn(const pmix_proc_t *proc,
  63                                           const pmix_info_t info[], size_t ninfo,
  64                                           pmix_modex_cbfunc_t cbfunc, void *cbdata);
  65 static pmix_status_t server_publish_fn(const pmix_proc_t *proc,
  66                                        const pmix_info_t info[], size_t ninfo,
  67                                        pmix_op_cbfunc_t cbfunc, void *cbdata);
  68 static pmix_status_t server_lookup_fn(const pmix_proc_t *proc,  char **keys,
  69                                       const pmix_info_t info[], size_t ninfo,
  70                                       pmix_lookup_cbfunc_t cbfunc, void *cbdata);
  71 static pmix_status_t server_unpublish_fn(const pmix_proc_t *proc, char **keys,
  72                                          const pmix_info_t info[], size_t ninfo,
  73                                          pmix_op_cbfunc_t cbfunc, void *cbdata);
  74 static pmix_status_t server_spawn_fn(const pmix_proc_t *proc,
  75                                      const pmix_info_t job_info[], size_t ninfo,
  76                                      const pmix_app_t apps[], size_t napps,
  77                                      pmix_spawn_cbfunc_t cbfunc, void *cbdata);
  78 static pmix_status_t server_connect_fn(const pmix_proc_t procs[], size_t nprocs,
  79                                        const pmix_info_t info[], size_t ninfo,
  80                                        pmix_op_cbfunc_t cbfunc, void *cbdata);
  81 static pmix_status_t server_disconnect_fn(const pmix_proc_t procs[], size_t nprocs,
  82                                           const pmix_info_t info[], size_t ninfo,
  83                                           pmix_op_cbfunc_t cbfunc, void *cbdata);
  84 static pmix_status_t server_register_events(pmix_status_t *codes, size_t ncodes,
  85                                             const pmix_info_t info[], size_t ninfo,
  86                                             pmix_op_cbfunc_t cbfunc, void *cbdata);
  87 static pmix_status_t server_deregister_events(pmix_status_t *codes, size_t ncodes,
  88                                               pmix_op_cbfunc_t cbfunc, void *cbdata);
  89 static pmix_status_t server_notify_event(pmix_status_t code,
  90                                          const pmix_proc_t *source,
  91                                          pmix_data_range_t range,
  92                                          pmix_info_t info[], size_t ninfo,
  93                                          pmix_op_cbfunc_t cbfunc, void *cbdata);
  94 static pmix_status_t server_query(pmix_proc_t *proct,
  95                                   pmix_query_t *queryies, size_t nqueries,
  96                                   pmix_info_cbfunc_t cbfunc,
  97                                   void *cbdata);
  98 static void server_tool_connection(pmix_info_t *info, size_t ninfo,
  99                                    pmix_tool_connection_cbfunc_t cbfunc,
 100                                    void *cbdata);
 101 static void server_log(const pmix_proc_t *client,
 102                        const pmix_info_t data[], size_t ndata,
 103                        const pmix_info_t directives[], size_t ndirs,
 104                        pmix_op_cbfunc_t cbfunc, void *cbdata);
 105 
 106 static pmix_status_t server_allocate(const pmix_proc_t *client,
 107                                      pmix_alloc_directive_t directive,
 108                                      const pmix_info_t data[], size_t ndata,
 109                                      pmix_info_cbfunc_t cbfunc, void *cbdata);
 110 
 111 static pmix_status_t server_job_control(const pmix_proc_t *requestor,
 112                                         const pmix_proc_t targets[], size_t ntargets,
 113                                         const pmix_info_t directives[], size_t ndirs,
 114                                         pmix_info_cbfunc_t cbfunc, void *cbdata);
 115 
 116 pmix_server_module_t mymodule = {
 117     .client_connected = server_client_connected_fn,
 118     .client_finalized = server_client_finalized_fn,
 119     .abort = server_abort_fn,
 120     .fence_nb = server_fencenb_fn,
 121     .direct_modex = server_dmodex_req_fn,
 122     .publish = server_publish_fn,
 123     .lookup = server_lookup_fn,
 124     .unpublish = server_unpublish_fn,
 125     .spawn = server_spawn_fn,
 126     .connect = server_connect_fn,
 127     .disconnect = server_disconnect_fn,
 128     .register_events = server_register_events,
 129     .deregister_events = server_deregister_events,
 130     .notify_event = server_notify_event,
 131     .query = server_query,
 132     .tool_connected = server_tool_connection,
 133     .log = server_log,
 134     .allocate = server_allocate,
 135     .job_control = server_job_control
 136     /* we do not support monitoring, but use the
 137      * PMIx internal monitoring capability */
 138 };
 139 
 140 opal_pmix_server_module_t *host_module = NULL;
 141 
 142 
 143 static void opal_opcbfunc(int status, void *cbdata)
 144 {
 145     ext2x_opalcaddy_t *opalcaddy = (ext2x_opalcaddy_t*)cbdata;
 146 
 147     OPAL_ACQUIRE_OBJECT(opalcaddy);
 148     if (NULL != opalcaddy->opcbfunc) {
 149         opalcaddy->opcbfunc(ext2x_convert_opalrc(status), opalcaddy->cbdata);
 150     }
 151     OBJ_RELEASE(opalcaddy);
 152 }
 153 
 154 static pmix_status_t server_client_connected_fn(const pmix_proc_t *p, void *server_object,
 155                                                 pmix_op_cbfunc_t cbfunc, void *cbdata)
 156 {
 157     int rc;
 158     opal_process_name_t proc;
 159     ext2x_opalcaddy_t *opalcaddy;
 160 
 161     if (NULL == host_module || NULL == host_module->client_connected) {
 162         return PMIX_SUCCESS;
 163     }
 164 
 165     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
 166     opalcaddy->opcbfunc = cbfunc;
 167     opalcaddy->cbdata = cbdata;
 168 
 169     /* convert the nspace/rank to an opal_process_name_t */
 170     if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&proc.jobid, p->nspace))) {
 171         return ext2x_convert_opalrc(rc);
 172     }
 173     proc.vpid = ext2x_convert_rank(p->rank);
 174 
 175     /* pass it up */
 176     rc = host_module->client_connected(&proc, server_object,
 177                                        opal_opcbfunc, opalcaddy);
 178     return ext2x_convert_opalrc(rc);
 179 }
 180 
 181 static pmix_status_t server_client_finalized_fn(const pmix_proc_t *p, void* server_object,
 182                                                 pmix_op_cbfunc_t cbfunc, void *cbdata)
 183 {
 184     int rc;
 185     ext2x_opalcaddy_t *opalcaddy;
 186     opal_process_name_t proc;
 187 
 188     if (NULL == host_module || NULL == host_module->client_finalized) {
 189         return PMIX_SUCCESS;
 190     }
 191 
 192     /* convert the nspace/rank to an opal_process_name_t */
 193     if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&proc.jobid, p->nspace))) {
 194         return ext2x_convert_opalrc(rc);
 195     }
 196     proc.vpid = ext2x_convert_rank(p->rank);
 197 
 198     /* setup the caddy */
 199     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
 200     opalcaddy->opcbfunc = cbfunc;
 201     opalcaddy->cbdata = cbdata;
 202 
 203     /* pass it up */
 204     opal_output_verbose(3, opal_pmix_base_framework.framework_output,
 205                         "%s CLIENT %s FINALIZED",
 206                         OPAL_NAME_PRINT(OPAL_PROC_MY_NAME),
 207                         OPAL_NAME_PRINT(proc));
 208     rc = host_module->client_finalized(&proc, server_object, opal_opcbfunc, opalcaddy);
 209     if (OPAL_SUCCESS != rc) {
 210         OBJ_RELEASE(opalcaddy);
 211     }
 212     return ext2x_convert_opalrc(rc);
 213 }
 214 
 215 static pmix_status_t server_abort_fn(const pmix_proc_t *p, void *server_object,
 216                                      int status, const char msg[],
 217                                      pmix_proc_t procs[], size_t nprocs,
 218                                      pmix_op_cbfunc_t cbfunc, void *cbdata)
 219 {
 220     size_t n;
 221     opal_namelist_t *nm;
 222     opal_process_name_t proc;
 223     int rc;
 224     ext2x_opalcaddy_t *opalcaddy;
 225 
 226     if (NULL == host_module || NULL == host_module->abort) {
 227         return PMIX_ERR_NOT_SUPPORTED;
 228     }
 229 
 230     /* convert the nspace/rank to an opal_process_name_t */
 231     if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&proc.jobid, p->nspace))) {
 232         return ext2x_convert_opalrc(rc);
 233     }
 234     proc.vpid = ext2x_convert_rank(p->rank);
 235 
 236     opal_output_verbose(3, opal_pmix_base_framework.framework_output,
 237                         "%s CLIENT %s CALLED ABORT",
 238                         OPAL_NAME_PRINT(OPAL_PROC_MY_NAME),
 239                         OPAL_NAME_PRINT(proc));
 240 
 241     /* setup the caddy */
 242     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
 243     opalcaddy->opcbfunc = cbfunc;
 244     opalcaddy->cbdata = cbdata;
 245 
 246     /* convert the array of pmix_proc_t to the list of procs */
 247     for (n=0; n < nprocs; n++) {
 248         nm = OBJ_NEW(opal_namelist_t);
 249         opal_list_append(&opalcaddy->procs, &nm->super);
 250         if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&nm->name.jobid, procs[n].nspace))) {
 251             OBJ_RELEASE(opalcaddy);
 252             return ext2x_convert_opalrc(rc);
 253         }
 254         nm->name.vpid = ext2x_convert_rank(procs[n].rank);
 255     }
 256 
 257     /* pass it up */
 258     rc = host_module->abort(&proc, server_object, status, msg,
 259                             &opalcaddy->procs, opal_opcbfunc, opalcaddy);
 260     if (OPAL_SUCCESS != rc) {
 261         OBJ_RELEASE(opalcaddy);
 262     }
 263     return ext2x_convert_opalrc(rc);
 264 }
 265 
 266 static void _data_release(void *cbdata)
 267 {
 268     ext2x_opalcaddy_t *opalcaddy = (ext2x_opalcaddy_t*)cbdata;
 269 
 270     if (NULL != opalcaddy->odmdxfunc) {
 271         opalcaddy->odmdxfunc(opalcaddy->ocbdata);
 272     }
 273     OBJ_RELEASE(opalcaddy);
 274 }
 275 
 276 static void opmdx_response(int status, const char *data, size_t sz, void *cbdata,
 277                            opal_pmix_release_cbfunc_t relcbfunc, void *relcbdata)
 278 {
 279     pmix_status_t rc;
 280     ext2x_opalcaddy_t *opalcaddy = (ext2x_opalcaddy_t*)cbdata;
 281     opal_ext2x_dmx_trkr_t *dmdx;
 282 
 283     rc = ext2x_convert_rc(status);
 284     if (NULL != opalcaddy->mdxcbfunc) {
 285         opalcaddy->odmdxfunc = relcbfunc;
 286         opalcaddy->ocbdata = relcbdata;
 287         opalcaddy->mdxcbfunc(rc, data, sz, opalcaddy->cbdata,
 288                              _data_release, opalcaddy);
 289         /* if we were collecting all data, then check for any pending
 290          * dmodx requests that we cached and notify them that the
 291          * data has arrived */
 292         OPAL_PMIX_ACQUIRE_THREAD(&opal_pmix_base.lock);
 293         while (NULL != (dmdx = (opal_ext2x_dmx_trkr_t*)opal_list_remove_first(&mca_pmix_ext2x_component.dmdx))) {
 294             dmdx->cbfunc(PMIX_SUCCESS, NULL, 0, dmdx->cbdata, NULL, NULL);
 295             OBJ_RELEASE(dmdx);
 296         }
 297         OPAL_PMIX_RELEASE_THREAD(&opal_pmix_base.lock);
 298     } else {
 299         OBJ_RELEASE(opalcaddy);
 300     }
 301 }
 302 
 303 static pmix_status_t server_fencenb_fn(const pmix_proc_t procs[], size_t nprocs,
 304                                        const pmix_info_t info[], size_t ninfo,
 305                                        char *data, size_t ndata,
 306                                        pmix_modex_cbfunc_t cbfunc, void *cbdata)
 307 {
 308     ext2x_opalcaddy_t *opalcaddy;
 309     size_t n;
 310     opal_namelist_t *nm;
 311     opal_value_t *iptr;
 312     int rc;
 313 
 314     opal_output_verbose(3, opal_pmix_base_framework.framework_output,
 315                         "%s FENCE CALLED", OPAL_NAME_PRINT(OPAL_PROC_MY_NAME));
 316 
 317     if (NULL == host_module || NULL == host_module->fence_nb) {
 318         return PMIX_ERR_NOT_SUPPORTED;
 319     }
 320     /* setup the caddy */
 321     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
 322     opalcaddy->mdxcbfunc = cbfunc;
 323     opalcaddy->cbdata = cbdata;
 324 
 325     /* convert the array of pmix_proc_t to the list of procs */
 326     for (n=0; n < nprocs; n++) {
 327         nm = OBJ_NEW(opal_namelist_t);
 328         opal_list_append(&opalcaddy->procs, &nm->super);
 329         if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&nm->name.jobid, procs[n].nspace))) {
 330             OBJ_RELEASE(opalcaddy);
 331             return ext2x_convert_opalrc(rc);
 332         }
 333         nm->name.vpid = ext2x_convert_rank(procs[n].rank);
 334     }
 335 
 336     /* convert the array of pmix_info_t to the list of info */
 337     for (n=0; n < ninfo; n++) {
 338         iptr = OBJ_NEW(opal_value_t);
 339         opal_list_append(&opalcaddy->info, &iptr->super);
 340         iptr->key = strdup(info[n].key);
 341         if (OPAL_SUCCESS != (rc = ext2x_value_unload(iptr, &info[n].value))) {
 342             OBJ_RELEASE(opalcaddy);
 343             return ext2x_convert_opalrc(rc);
 344         }
 345     }
 346 
 347     /* pass it up */
 348     rc = host_module->fence_nb(&opalcaddy->procs, &opalcaddy->info,
 349                                data, ndata, opmdx_response, opalcaddy);
 350     if (OPAL_SUCCESS != rc) {
 351         OBJ_RELEASE(opalcaddy);
 352     }
 353     return ext2x_convert_opalrc(rc);
 354 }
 355 
 356 static pmix_status_t server_dmodex_req_fn(const pmix_proc_t *p,
 357                                           const pmix_info_t info[], size_t ninfo,
 358                                           pmix_modex_cbfunc_t cbfunc, void *cbdata)
 359 {
 360     int rc;
 361     ext2x_opalcaddy_t *opalcaddy;
 362     opal_process_name_t proc;
 363     opal_value_t *iptr;
 364     size_t n;
 365     opal_ext2x_dmx_trkr_t *dmdx;
 366 
 367     if (NULL == host_module || NULL == host_module->direct_modex) {
 368         return PMIX_ERR_NOT_SUPPORTED;
 369     }
 370 
 371     /* convert the nspace/rank to an opal_process_name_t */
 372     if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&proc.jobid, p->nspace))) {
 373         return ext2x_convert_opalrc(rc);
 374     }
 375     proc.vpid = ext2x_convert_rank(p->rank);
 376 
 377     opal_output_verbose(3, opal_pmix_base_framework.framework_output,
 378                         "%s CLIENT %s CALLED DMODX",
 379                         OPAL_NAME_PRINT(OPAL_PROC_MY_NAME),
 380                         OPAL_NAME_PRINT(proc));
 381 
 382     /* setup the caddy */
 383     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
 384     opalcaddy->mdxcbfunc = cbfunc;
 385     opalcaddy->cbdata = cbdata;
 386 
 387     /* this function should only get called if we are in an async modex.
 388      * If we are also collecting data, then the fence_nb will eventually
 389      * complete and return all the required data down to the pmix
 390      * server beneath us. Thus, we only need to track the dmodex_req
 391      * and ensure that the release gets called once the data has
 392      * arrived - this will trigger the pmix server to tell the
 393      * client that the data is available */
 394     if (opal_pmix_base_async_modex && opal_pmix_collect_all_data) {
 395         OPAL_PMIX_ACQUIRE_THREAD(&opal_pmix_base.lock);
 396         dmdx = OBJ_NEW(opal_ext2x_dmx_trkr_t);
 397         dmdx->cbfunc = cbfunc;
 398         dmdx->cbdata = cbdata;
 399         opal_list_append(&mca_pmix_ext2x_component.dmdx, &dmdx->super);
 400         OPAL_PMIX_RELEASE_THREAD(&opal_pmix_base.lock);
 401         return PMIX_SUCCESS;
 402     }
 403 
 404     /* convert the array of pmix_info_t to the list of info */
 405     for (n=0; n < ninfo; n++) {
 406         iptr = OBJ_NEW(opal_value_t);
 407         opal_list_append(&opalcaddy->info, &iptr->super);
 408         iptr->key = strdup(info[n].key);
 409         if (OPAL_SUCCESS != (rc = ext2x_value_unload(iptr, &info[n].value))) {
 410             OBJ_RELEASE(opalcaddy);
 411             return ext2x_convert_opalrc(rc);
 412         }
 413     }
 414 
 415     /* pass it up */
 416     rc = host_module->direct_modex(&proc, &opalcaddy->info, opmdx_response, opalcaddy);
 417     if (OPAL_SUCCESS != rc && OPAL_ERR_IN_PROCESS != rc) {
 418         OBJ_RELEASE(opalcaddy);
 419     }
 420     if (OPAL_ERR_IN_PROCESS == rc) {
 421         rc = OPAL_SUCCESS;
 422     }
 423     return ext2x_convert_opalrc(rc);
 424 }
 425 
 426 static pmix_status_t server_publish_fn(const pmix_proc_t *p,
 427                                        const pmix_info_t info[], size_t ninfo,
 428                                        pmix_op_cbfunc_t cbfunc, void *cbdata)
 429 {
 430     int rc;
 431     size_t n;
 432     ext2x_opalcaddy_t *opalcaddy;
 433     opal_process_name_t proc;
 434     opal_value_t *oinfo;
 435 
 436     if (NULL == host_module || NULL == host_module->publish) {
 437         return PMIX_ERR_NOT_SUPPORTED;
 438     }
 439 
 440    /* convert the nspace/rank to an opal_process_name_t */
 441     if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&proc.jobid, p->nspace))) {
 442         return ext2x_convert_opalrc(rc);
 443     }
 444     proc.vpid = ext2x_convert_rank(p->rank);
 445 
 446     opal_output_verbose(3, opal_pmix_base_framework.framework_output,
 447                         "%s CLIENT %s CALLED PUBLISH",
 448                         OPAL_NAME_PRINT(OPAL_PROC_MY_NAME),
 449                         OPAL_NAME_PRINT(proc));
 450 
 451     /* setup the caddy */
 452     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
 453     opalcaddy->opcbfunc = cbfunc;
 454     opalcaddy->cbdata = cbdata;
 455 
 456     /* convert the info array */
 457     for (n=0; n < ninfo; n++) {
 458         oinfo = OBJ_NEW(opal_value_t);
 459         opal_list_append(&opalcaddy->info, &oinfo->super);
 460         oinfo->key = strdup(info[n].key);
 461         if (OPAL_SUCCESS != (rc = ext2x_value_unload(oinfo, &info[n].value))) {
 462             OBJ_RELEASE(opalcaddy);
 463             return ext2x_convert_opalrc(rc);
 464         }
 465     }
 466 
 467     /* pass it up */
 468     rc = host_module->publish(&proc, &opalcaddy->info, opal_opcbfunc, opalcaddy);
 469     if (OPAL_SUCCESS != rc) {
 470         OBJ_RELEASE(opalcaddy);
 471     }
 472 
 473     return ext2x_convert_opalrc(rc);
 474 }
 475 
 476 static void opal_lkupcbfunc(int status,
 477                             opal_list_t *data,
 478                             void *cbdata)
 479 {
 480     ext2x_opalcaddy_t *opalcaddy = (ext2x_opalcaddy_t*)cbdata;
 481     pmix_status_t rc;
 482     pmix_pdata_t *d=NULL;
 483     size_t nd=0, n;
 484     opal_pmix_pdata_t *p;
 485 
 486     if (NULL != opalcaddy->lkupcbfunc) {
 487         rc = ext2x_convert_opalrc(status);
 488         /* convert any returned data */
 489         if (NULL != data) {
 490             nd = opal_list_get_size(data);
 491             PMIX_PDATA_CREATE(d, nd);
 492             n=0;
 493             OPAL_LIST_FOREACH(p, data, opal_pmix_pdata_t) {
 494                 /* convert the jobid */
 495                 (void)opal_snprintf_jobid(d[n].proc.nspace, PMIX_MAX_NSLEN, p->proc.jobid);
 496                 d[n].proc.rank = ext2x_convert_opalrank(p->proc.vpid);
 497                 (void)opal_string_copy(d[n].key, p->value.key, PMIX_MAX_KEYLEN);
 498                 ext2x_value_load(&d[n].value, &p->value);
 499             }
 500         }
 501         opalcaddy->lkupcbfunc(rc, d, nd, opalcaddy->cbdata);
 502         PMIX_PDATA_FREE(d, nd);
 503     }
 504     OBJ_RELEASE(opalcaddy);
 505 }
 506 
 507 static pmix_status_t server_lookup_fn(const pmix_proc_t *p, char **keys,
 508                                       const pmix_info_t info[], size_t ninfo,
 509                                       pmix_lookup_cbfunc_t cbfunc, void *cbdata)
 510 {
 511     int rc;
 512     ext2x_opalcaddy_t *opalcaddy;
 513     opal_process_name_t proc;
 514     opal_value_t *iptr;
 515     size_t n;
 516 
 517     if (NULL == host_module || NULL == host_module->lookup) {
 518         return PMIX_ERR_NOT_SUPPORTED;
 519     }
 520 
 521     /* convert the nspace/rank to an opal_process_name_t */
 522     if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&proc.jobid, p->nspace))) {
 523         return ext2x_convert_opalrc(rc);
 524     }
 525     proc.vpid = ext2x_convert_rank(p->rank);
 526 
 527     opal_output_verbose(3, opal_pmix_base_framework.framework_output,
 528                         "%s CLIENT %s CALLED LOOKUP",
 529                         OPAL_NAME_PRINT(OPAL_PROC_MY_NAME),
 530                         OPAL_NAME_PRINT(proc));
 531 
 532     /* setup the caddy */
 533     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
 534     opalcaddy->lkupcbfunc = cbfunc;
 535     opalcaddy->cbdata = cbdata;
 536 
 537     /* convert the array of pmix_info_t to the list of info */
 538     for (n=0; n < ninfo; n++) {
 539         iptr = OBJ_NEW(opal_value_t);
 540         opal_list_append(&opalcaddy->info, &iptr->super);
 541         iptr->key = strdup(info[n].key);
 542         if (OPAL_SUCCESS != (rc = ext2x_value_unload(iptr, &info[n].value))) {
 543             OBJ_RELEASE(opalcaddy);
 544             return ext2x_convert_opalrc(rc);
 545         }
 546     }
 547 
 548     /* pass it up */
 549     rc = host_module->lookup(&proc, keys, &opalcaddy->info, opal_lkupcbfunc, opalcaddy);
 550     if (OPAL_SUCCESS != rc) {
 551         OBJ_RELEASE(opalcaddy);
 552     }
 553 
 554     return ext2x_convert_opalrc(rc);
 555 }
 556 
 557 
 558 static pmix_status_t server_unpublish_fn(const pmix_proc_t *p, char **keys,
 559                                          const pmix_info_t info[], size_t ninfo,
 560                                          pmix_op_cbfunc_t cbfunc, void *cbdata)
 561 {
 562     int rc;
 563     ext2x_opalcaddy_t *opalcaddy;
 564     opal_process_name_t proc;
 565     opal_value_t *iptr;
 566     size_t n;
 567 
 568     if (NULL == host_module || NULL == host_module->unpublish) {
 569         return PMIX_SUCCESS;
 570     }
 571 
 572     /* convert the nspace/rank to an opal_process_name_t */
 573     if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&proc.jobid, p->nspace))) {
 574         return ext2x_convert_opalrc(rc);
 575     }
 576     proc.vpid = ext2x_convert_rank(p->rank);
 577 
 578     opal_output_verbose(3, opal_pmix_base_framework.framework_output,
 579                         "%s CLIENT %s CALLED UNPUBLISH",
 580                         OPAL_NAME_PRINT(OPAL_PROC_MY_NAME),
 581                         OPAL_NAME_PRINT(proc));
 582 
 583     /* setup the caddy */
 584     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
 585     opalcaddy->opcbfunc = cbfunc;
 586     opalcaddy->cbdata = cbdata;
 587 
 588     /* convert the array of pmix_info_t to the list of info */
 589     for (n=0; n < ninfo; n++) {
 590         iptr = OBJ_NEW(opal_value_t);
 591         opal_list_append(&opalcaddy->info, &iptr->super);
 592         iptr->key = strdup(info[n].key);
 593         if (OPAL_SUCCESS != (rc = ext2x_value_unload(iptr, &info[n].value))) {
 594             OBJ_RELEASE(opalcaddy);
 595             return ext2x_convert_opalrc(rc);
 596         }
 597     }
 598 
 599     /* pass it up */
 600     rc = host_module->unpublish(&proc, keys, &opalcaddy->info, opal_opcbfunc, opalcaddy);
 601     if (OPAL_SUCCESS != rc) {
 602         OBJ_RELEASE(opalcaddy);
 603     }
 604 
 605     return ext2x_convert_opalrc(rc);
 606 }
 607 
 608 static void opal_spncbfunc(int status, opal_jobid_t jobid, void *cbdata)
 609 {
 610     ext2x_opalcaddy_t *opalcaddy = (ext2x_opalcaddy_t*)cbdata;
 611     pmix_status_t rc;
 612     char nspace[PMIX_MAX_NSLEN];
 613 
 614     if (NULL != opalcaddy->spwncbfunc) {
 615         rc = ext2x_convert_opalrc(status);
 616         /* convert the jobid */
 617         (void)opal_snprintf_jobid(nspace, PMIX_MAX_NSLEN, jobid);
 618         opalcaddy->spwncbfunc(rc, nspace, opalcaddy->cbdata);
 619     }
 620     OBJ_RELEASE(opalcaddy);
 621 }
 622 
 623 static pmix_status_t server_spawn_fn(const pmix_proc_t *p,
 624                                      const pmix_info_t job_info[], size_t ninfo,
 625                                      const pmix_app_t apps[], size_t napps,
 626                                      pmix_spawn_cbfunc_t cbfunc, void *cbdata)
 627 {
 628     ext2x_opalcaddy_t *opalcaddy;
 629     opal_process_name_t proc;
 630     opal_pmix_app_t *app;
 631     opal_value_t *oinfo;
 632     size_t k, n;
 633     int rc;
 634 
 635     if (NULL == host_module || NULL == host_module->spawn) {
 636         return PMIX_ERR_NOT_SUPPORTED;
 637     }
 638 
 639     /* convert the nspace/rank to an opal_process_name_t */
 640     if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&proc.jobid, p->nspace))) {
 641         return ext2x_convert_opalrc(rc);
 642     }
 643     proc.vpid = ext2x_convert_rank(p->rank);
 644 
 645     /* setup the caddy */
 646     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
 647     opalcaddy->spwncbfunc = cbfunc;
 648     opalcaddy->cbdata = cbdata;
 649 
 650     /* convert the job info */
 651     for (k=0; k < ninfo; k++) {
 652         oinfo = OBJ_NEW(opal_value_t);
 653         opal_list_append(&opalcaddy->info, &oinfo->super);
 654         oinfo->key = strdup(job_info[k].key);
 655         if (OPAL_SUCCESS != (rc = ext2x_value_unload(oinfo, &job_info[k].value))) {
 656             OBJ_RELEASE(opalcaddy);
 657             return ext2x_convert_opalrc(rc);
 658         }
 659     }
 660 
 661     /* convert the apps */
 662     for (n=0; n < napps; n++) {
 663         app = OBJ_NEW(opal_pmix_app_t);
 664         opal_list_append(&opalcaddy->apps, &app->super);
 665         if (NULL != apps[n].cmd) {
 666             app->cmd = strdup(apps[n].cmd);
 667         }
 668         if (NULL != apps[n].argv) {
 669             app->argv = opal_argv_copy(apps[n].argv);
 670         }
 671         if (NULL != apps[n].env) {
 672             app->env = opal_argv_copy(apps[n].env);
 673         }
 674         if (NULL != apps[n].cwd) {
 675             app->cwd = strdup(apps[n].cwd);
 676         }
 677         app->maxprocs = apps[n].maxprocs;
 678         for (k=0; k < apps[n].ninfo; k++) {
 679             oinfo = OBJ_NEW(opal_value_t);
 680             opal_list_append(&app->info, &oinfo->super);
 681             oinfo->key = strdup(apps[n].info[k].key);
 682             if (OPAL_SUCCESS != (rc = ext2x_value_unload(oinfo, &apps[n].info[k].value))) {
 683                 OBJ_RELEASE(opalcaddy);
 684                 return ext2x_convert_opalrc(rc);
 685             }
 686         }
 687     }
 688 
 689     /* pass it up */
 690     rc = host_module->spawn(&proc, &opalcaddy->info, &opalcaddy->apps, opal_spncbfunc, opalcaddy);
 691     if (OPAL_SUCCESS != rc) {
 692         OPAL_ERROR_LOG(rc);
 693         OBJ_RELEASE(opalcaddy);
 694     }
 695 
 696     return ext2x_convert_opalrc(rc);
 697 }
 698 
 699 
 700 static pmix_status_t server_connect_fn(const pmix_proc_t procs[], size_t nprocs,
 701                                        const pmix_info_t info[], size_t ninfo,
 702                                        pmix_op_cbfunc_t cbfunc, void *cbdata)
 703 {
 704     int rc;
 705     ext2x_opalcaddy_t *opalcaddy;
 706     opal_namelist_t *nm;
 707     size_t n;
 708     opal_value_t *oinfo;
 709 
 710     if (NULL == host_module || NULL == host_module->connect) {
 711         return PMIX_ERR_NOT_SUPPORTED;
 712     }
 713 
 714     /* setup the caddy */
 715     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
 716     opalcaddy->opcbfunc = cbfunc;
 717     opalcaddy->cbdata = cbdata;
 718 
 719     /* convert the array of pmix_proc_t to the list of procs */
 720     for (n=0; n < nprocs; n++) {
 721         nm = OBJ_NEW(opal_namelist_t);
 722         opal_list_append(&opalcaddy->procs, &nm->super);
 723         if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&nm->name.jobid, procs[n].nspace))) {
 724             OBJ_RELEASE(opalcaddy);
 725             return ext2x_convert_opalrc(rc);
 726         }
 727         nm->name.vpid = ext2x_convert_rank(procs[n].rank);
 728     }
 729 
 730     /* convert the info */
 731     for (n=0; n < ninfo; n++) {
 732         oinfo = OBJ_NEW(opal_value_t);
 733         opal_list_append(&opalcaddy->info, &oinfo->super);
 734         oinfo->key = strdup(info[n].key);
 735         if (OPAL_SUCCESS != (rc = ext2x_value_unload(oinfo, &info[n].value))) {
 736             OBJ_RELEASE(opalcaddy);
 737             return ext2x_convert_opalrc(rc);
 738         }
 739     }
 740 
 741     /* pass it up */
 742     rc = host_module->connect(&opalcaddy->procs, &opalcaddy->info, opal_opcbfunc, opalcaddy);
 743     if (OPAL_SUCCESS != rc) {
 744         OBJ_RELEASE(opalcaddy);
 745     }
 746 
 747     return ext2x_convert_opalrc(rc);
 748 }
 749 
 750 
 751 static pmix_status_t server_disconnect_fn(const pmix_proc_t procs[], size_t nprocs,
 752                                           const pmix_info_t info[], size_t ninfo,
 753                                           pmix_op_cbfunc_t cbfunc, void *cbdata)
 754 {
 755     int rc;
 756     ext2x_opalcaddy_t *opalcaddy;
 757     opal_namelist_t *nm;
 758     size_t n;
 759     opal_value_t *oinfo;
 760 
 761     if (NULL == host_module || NULL == host_module->disconnect) {
 762         return PMIX_ERR_NOT_SUPPORTED;
 763     }
 764 
 765     /* setup the caddy */
 766     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
 767     opalcaddy->opcbfunc = cbfunc;
 768     opalcaddy->cbdata = cbdata;
 769 
 770     /* convert the array of pmix_proc_t to the list of procs */
 771     for (n=0; n < nprocs; n++) {
 772         nm = OBJ_NEW(opal_namelist_t);
 773         opal_list_append(&opalcaddy->procs, &nm->super);
 774         if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&nm->name.jobid, procs[n].nspace))) {
 775             OBJ_RELEASE(opalcaddy);
 776             return ext2x_convert_opalrc(rc);
 777         }
 778         nm->name.vpid = ext2x_convert_rank(procs[n].rank);
 779     }
 780 
 781     /* convert the info */
 782     for (n=0; n < ninfo; n++) {
 783         oinfo = OBJ_NEW(opal_value_t);
 784         opal_list_append(&opalcaddy->info, &oinfo->super);
 785         oinfo->key = strdup(info[n].key);
 786         if (OPAL_SUCCESS != (rc = ext2x_value_unload(oinfo, &info[n].value))) {
 787             OBJ_RELEASE(opalcaddy);
 788             return ext2x_convert_opalrc(rc);
 789         }
 790     }
 791 
 792     /* pass it up */
 793     rc = host_module->disconnect(&opalcaddy->procs, &opalcaddy->info, opal_opcbfunc, opalcaddy);
 794     if (OPAL_SUCCESS != rc) {
 795         OBJ_RELEASE(opalcaddy);
 796     }
 797 
 798     return ext2x_convert_opalrc(rc);
 799 }
 800 
 801 static pmix_status_t server_register_events(pmix_status_t *codes, size_t ncodes,
 802                                             const pmix_info_t info[], size_t ninfo,
 803                                             pmix_op_cbfunc_t cbfunc, void *cbdata)
 804 {
 805     ext2x_opalcaddy_t *opalcaddy;
 806     size_t n;
 807     opal_value_t *oinfo;
 808     int rc;
 809 
 810     opal_output_verbose(3, opal_pmix_base_framework.framework_output,
 811                         "%s REGISTER EVENTS",
 812                         OPAL_NAME_PRINT(OPAL_PROC_MY_NAME));
 813 
 814     /* setup the caddy */
 815     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
 816     opalcaddy->opcbfunc = cbfunc;
 817     opalcaddy->cbdata = cbdata;
 818 
 819     /* convert the info */
 820     for (n=0; n < ninfo; n++) {
 821         oinfo = OBJ_NEW(opal_value_t);
 822         opal_list_append(&opalcaddy->info, &oinfo->super);
 823         oinfo->key = strdup(info[n].key);
 824         if (OPAL_SUCCESS != (rc = ext2x_value_unload(oinfo, &info[n].value))) {
 825             OBJ_RELEASE(opalcaddy);
 826             return ext2x_convert_opalrc(rc);
 827         }
 828     }
 829 
 830     /* pass it up */
 831     rc = host_module->register_events(&opalcaddy->info, opal_opcbfunc, opalcaddy);
 832     if (OPAL_SUCCESS != rc) {
 833         OBJ_RELEASE(opalcaddy);
 834     }
 835 
 836     return ext2x_convert_opalrc(rc);
 837 }
 838 
 839 static pmix_status_t server_deregister_events(pmix_status_t *codes, size_t ncodes,
 840                                               pmix_op_cbfunc_t cbfunc, void *cbdata)
 841 {
 842     opal_output_verbose(3, opal_pmix_base_framework.framework_output,
 843                         "%s DEREGISTER EVENTS", OPAL_NAME_PRINT(OPAL_PROC_MY_NAME));
 844 
 845     return PMIX_ERR_NOT_SUPPORTED;
 846 }
 847 
 848 static pmix_status_t server_notify_event(pmix_status_t code,
 849                                          const pmix_proc_t *source,
 850                                          pmix_data_range_t range,
 851                                          pmix_info_t info[], size_t ninfo,
 852                                          pmix_op_cbfunc_t cbfunc, void *cbdata)
 853 {
 854     ext2x_opalcaddy_t *opalcaddy;
 855     opal_process_name_t src;
 856     size_t n;
 857     opal_value_t *oinfo;
 858     int rc, status;
 859 
 860     if (NULL == host_module || NULL == host_module->notify_event) {
 861         return PMIX_ERR_NOT_SUPPORTED;
 862     }
 863 
 864     /* setup the caddy */
 865     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
 866     opalcaddy->opcbfunc = cbfunc;
 867     opalcaddy->cbdata = cbdata;
 868 
 869     /* convert the code */
 870     status = ext2x_convert_rc(code);
 871 
 872     /* convert the source */
 873     if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&src.jobid, source->nspace))) {
 874         OBJ_RELEASE(opalcaddy);
 875         return ext2x_convert_opalrc(rc);
 876     }
 877     src.vpid = ext2x_convert_rank(source->rank);
 878 
 879     opal_output_verbose(3, opal_pmix_base_framework.framework_output,
 880                         "%s CLIENT %s CALLED NOTIFY",
 881                         OPAL_NAME_PRINT(OPAL_PROC_MY_NAME),
 882                         OPAL_NAME_PRINT(src));
 883 
 884     /* ignore the range for now */
 885 
 886     /* convert the info */
 887     for (n=0; n < ninfo; n++) {
 888         oinfo = OBJ_NEW(opal_value_t);
 889         opal_list_append(&opalcaddy->info, &oinfo->super);
 890         oinfo->key = strdup(info[n].key);
 891         if (OPAL_SUCCESS != (rc = ext2x_value_unload(oinfo, &info[n].value))) {
 892             OBJ_RELEASE(opalcaddy);
 893             return ext2x_convert_opalrc(rc);
 894         }
 895     }
 896 
 897     /* send it upstairs */
 898     if (OPAL_SUCCESS != (rc = host_module->notify_event(status, &src, &opalcaddy->info,
 899                                                         opal_opcbfunc, opalcaddy))) {
 900         OBJ_RELEASE(opalcaddy);
 901     }
 902     return ext2x_convert_opalrc(rc);
 903 }
 904 
 905 static void _info_rel(void *cbdata)
 906 {
 907     ext2x_opcaddy_t *pcaddy = (ext2x_opcaddy_t*)cbdata;
 908 
 909     OBJ_RELEASE(pcaddy);
 910 }
 911 static void info_cbfunc(int status,
 912                         opal_list_t *info,
 913                         void *cbdata,
 914                         opal_pmix_release_cbfunc_t release_fn,
 915                         void *release_cbdata)
 916 {
 917     ext2x_opalcaddy_t *opalcaddy = (ext2x_opalcaddy_t*)cbdata;
 918     ext2x_opcaddy_t *pcaddy;
 919     opal_value_t *kv;
 920     size_t n;
 921 
 922     pcaddy = OBJ_NEW(ext2x_opcaddy_t);
 923 
 924     /* convert the status */
 925     pcaddy->status = ext2x_convert_opalrc(status);
 926 
 927     /* convert the list to a pmix_info_t array */
 928     if (NULL != info && 0 < (pcaddy->ninfo = opal_list_get_size(info))) {
 929         PMIX_INFO_CREATE(pcaddy->info, pcaddy->ninfo);
 930         n = 0;
 931         OPAL_LIST_FOREACH(kv, info, opal_value_t) {
 932             (void)opal_string_copy(pcaddy->info[n].key, kv->key, PMIX_MAX_KEYLEN);
 933             ext2x_value_load(&pcaddy->info[n].value, kv);
 934         }
 935     }
 936     /* we are done with the incoming data */
 937     if (NULL != release_fn) {
 938         release_fn(release_cbdata);
 939     }
 940 
 941     /* provide the answer downward */
 942     if (NULL != opalcaddy->infocbfunc) {
 943         opalcaddy->infocbfunc(pcaddy->status, pcaddy->info, pcaddy->ninfo,
 944                               opalcaddy->cbdata, _info_rel, pcaddy);
 945     }
 946     OBJ_RELEASE(opalcaddy);
 947 }
 948 
 949 static pmix_status_t server_query(pmix_proc_t *proct,
 950                                   pmix_query_t *queries, size_t nqueries,
 951                                   pmix_info_cbfunc_t cbfunc,
 952                                   void *cbdata)
 953 {
 954     ext2x_opalcaddy_t *opalcaddy;
 955     opal_process_name_t requestor;
 956     int rc;
 957     size_t n, m;
 958     opal_pmix_query_t *q;
 959     opal_value_t *oinfo;
 960 
 961     if (NULL == host_module || NULL == host_module->query) {
 962         return PMIX_ERR_NOT_SUPPORTED;
 963     }
 964 
 965     /* setup the caddy */
 966     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
 967     opalcaddy->infocbfunc = cbfunc;
 968     opalcaddy->cbdata = cbdata;
 969 
 970     /* convert the requestor */
 971     if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&requestor.jobid, proct->nspace))) {
 972         OBJ_RELEASE(opalcaddy);
 973         return ext2x_convert_opalrc(rc);
 974     }
 975     requestor.vpid = ext2x_convert_rank(proct->rank);
 976 
 977     opal_output_verbose(3, opal_pmix_base_framework.framework_output,
 978                         "%s CLIENT %s CALLED QUERY",
 979                         OPAL_NAME_PRINT(OPAL_PROC_MY_NAME),
 980                         OPAL_NAME_PRINT(requestor));
 981 
 982     /* convert the queries */
 983     for (n=0; n < nqueries; n++) {
 984         q = OBJ_NEW(opal_pmix_query_t);
 985         /* we "borrow" the info field of the caddy as we and the
 986          * server function both agree on what will be there */
 987         opal_list_append(&opalcaddy->info, &q->super);
 988         q->keys = opal_argv_copy(queries[n].keys);
 989         for (m=0; m < queries[n].nqual; m++) {
 990             oinfo = OBJ_NEW(opal_value_t);
 991             opal_list_append(&q->qualifiers, &oinfo->super);
 992             oinfo->key = strdup(queries[n].qualifiers[m].key);
 993             if (OPAL_SUCCESS != (rc = ext2x_value_unload(oinfo, &queries[n].qualifiers[m].value))) {
 994                 OBJ_RELEASE(opalcaddy);
 995                 return ext2x_convert_opalrc(rc);
 996             }
 997         }
 998     }
 999 
1000     /* pass the call upwards */
1001     if (OPAL_SUCCESS != (rc = host_module->query(&requestor,
1002                                                  &opalcaddy->info,
1003                                                  info_cbfunc, opalcaddy))) {
1004         OBJ_RELEASE(opalcaddy);
1005     }
1006 
1007     return ext2x_convert_opalrc(rc);
1008 }
1009 
1010 static void toolcbfunc(int status,
1011                        opal_process_name_t proc,
1012                        void *cbdata)
1013 {
1014     ext2x_opalcaddy_t *opalcaddy = (ext2x_opalcaddy_t*)cbdata;
1015     pmix_status_t rc;
1016     pmix_proc_t p;
1017     opal_ext2x_jobid_trkr_t *job;
1018 
1019     /* convert the status */
1020     rc = ext2x_convert_opalrc(status);
1021 
1022     memset(&p, 0, sizeof(pmix_proc_t));
1023     if (OPAL_SUCCESS == status) {
1024         /* convert the process name */
1025         (void)opal_snprintf_jobid(p.nspace, PMIX_MAX_NSLEN, proc.jobid);
1026         p.rank = ext2x_convert_opalrank(proc.vpid);
1027         /* store this job in our list of known nspaces */
1028         job = OBJ_NEW(opal_ext2x_jobid_trkr_t);
1029         (void)opal_string_copy(job->nspace, p.nspace, PMIX_MAX_NSLEN);
1030         job->jobid = proc.jobid;
1031         OPAL_PMIX_ACQUIRE_THREAD(&opal_pmix_base.lock);
1032         opal_list_append(&mca_pmix_ext2x_component.jobids, &job->super);
1033         OPAL_PMIX_RELEASE_THREAD(&opal_pmix_base.lock);
1034     }
1035 
1036     /* pass it down */
1037     if (NULL != opalcaddy->toolcbfunc) {
1038         opalcaddy->toolcbfunc(rc, &p, opalcaddy->cbdata);
1039     }
1040     OBJ_RELEASE(opalcaddy);
1041 }
1042 
1043 static void server_tool_connection(pmix_info_t *info, size_t ninfo,
1044                                    pmix_tool_connection_cbfunc_t cbfunc,
1045                                    void *cbdata)
1046 {
1047     ext2x_opalcaddy_t *opalcaddy;
1048     size_t n;
1049     opal_value_t *oinfo;
1050     int rc;
1051     pmix_status_t err;
1052 
1053     /* setup the caddy */
1054     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
1055     opalcaddy->toolcbfunc = cbfunc;
1056     opalcaddy->cbdata = cbdata;
1057 
1058     /* convert the info */
1059     for (n=0; n < ninfo; n++) {
1060         oinfo = OBJ_NEW(opal_value_t);
1061         opal_list_append(&opalcaddy->info, &oinfo->super);
1062         oinfo->key = strdup(info[n].key);
1063         if (OPAL_SUCCESS != (rc = ext2x_value_unload(oinfo, &info[n].value))) {
1064             OBJ_RELEASE(opalcaddy);
1065             err = ext2x_convert_opalrc(rc);
1066             if (NULL != cbfunc) {
1067                 cbfunc(err, NULL, cbdata);
1068             }
1069         }
1070     }
1071 
1072     /* pass it up */
1073     host_module->tool_connected(&opalcaddy->info, toolcbfunc, opalcaddy);
1074 }
1075 
1076 static void server_log(const pmix_proc_t *proct,
1077                        const pmix_info_t data[], size_t ndata,
1078                        const pmix_info_t directives[], size_t ndirs,
1079                        pmix_op_cbfunc_t cbfunc, void *cbdata)
1080 {
1081     ext2x_opalcaddy_t *opalcaddy;
1082     opal_process_name_t requestor;
1083     int rc;
1084     size_t n;
1085     opal_value_t *oinfo;
1086     pmix_status_t ret;
1087 
1088     if (NULL == host_module || NULL == host_module->log) {
1089         if (NULL != cbfunc) {
1090             cbfunc(PMIX_ERR_NOT_SUPPORTED, cbdata);
1091         }
1092         return;
1093     }
1094 
1095     /* setup the caddy */
1096     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
1097     opalcaddy->opcbfunc = cbfunc;
1098     opalcaddy->cbdata = cbdata;
1099 
1100     /* convert the requestor */
1101     if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&requestor.jobid, proct->nspace))) {
1102     OBJ_RELEASE(opalcaddy);
1103     ret = ext2x_convert_opalrc(rc);
1104         if (NULL != cbfunc) {
1105             cbfunc(ret, cbdata);
1106         }
1107         return;
1108     }
1109     requestor.vpid = ext2x_convert_rank(proct->rank);
1110 
1111     /* convert the data */
1112     for (n=0; n < ndata; n++) {
1113         oinfo = OBJ_NEW(opal_value_t);
1114         oinfo->key = strdup(data[n].key);
1115         /* we "borrow" the info field of the caddy as we and the
1116          * server function both agree on what will be there */
1117         opal_list_append(&opalcaddy->info, &oinfo->super);
1118         if (OPAL_SUCCESS != (rc = ext2x_value_unload(oinfo, &data[n].value))) {
1119             OBJ_RELEASE(opalcaddy);
1120             ret = ext2x_convert_opalrc(rc);
1121             if (NULL != cbfunc) {
1122                 cbfunc(ret, cbdata);
1123             }
1124             return;
1125         }
1126     }
1127 
1128     /* convert the directives */
1129     for (n=0; n < ndirs; n++) {
1130         oinfo = OBJ_NEW(opal_value_t);
1131         /* we "borrow" the apps field of the caddy as we and the
1132          * server function both agree on what will be there */
1133         opal_list_append(&opalcaddy->apps, &oinfo->super);
1134         if (OPAL_SUCCESS != (rc = ext2x_value_unload(oinfo, &directives[n].value))) {
1135             OBJ_RELEASE(opalcaddy);
1136             ret = ext2x_convert_opalrc(rc);
1137             if (NULL != cbfunc) {
1138                 cbfunc(ret, cbdata);
1139             }
1140             return;
1141         }
1142     }
1143 
1144     /* pass the call upwards */
1145     host_module->log(&requestor,
1146                      &opalcaddy->info,
1147                      &opalcaddy->apps,
1148                      opal_opcbfunc, opalcaddy);
1149 }
1150 
1151 static pmix_status_t server_allocate(const pmix_proc_t *proct,
1152                                      pmix_alloc_directive_t directive,
1153                                      const pmix_info_t data[], size_t ndata,
1154                                      pmix_info_cbfunc_t cbfunc, void *cbdata)
1155 {
1156     ext2x_opalcaddy_t *opalcaddy;
1157     opal_process_name_t requestor;
1158     int rc;
1159     size_t n;
1160     opal_value_t *oinfo;
1161     opal_pmix_alloc_directive_t odir;
1162 
1163     if (NULL == host_module || NULL == host_module->allocate) {
1164         return PMIX_ERR_NOT_SUPPORTED;
1165     }
1166 
1167     /* setup the caddy */
1168     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
1169     opalcaddy->infocbfunc = cbfunc;
1170     opalcaddy->cbdata = cbdata;
1171 
1172     /* convert the requestor */
1173     if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&requestor.jobid, proct->nspace))) {
1174         OBJ_RELEASE(opalcaddy);
1175         return ext2x_convert_opalrc(rc);
1176     }
1177     requestor.vpid = ext2x_convert_rank(proct->rank);
1178 
1179     /* convert the directive */
1180     odir = ext2x_convert_allocdir(directive);
1181 
1182     /* convert the data */
1183     for (n=0; n < ndata; n++) {
1184         oinfo = OBJ_NEW(opal_value_t);
1185         opal_list_append(&opalcaddy->info, &oinfo->super);
1186         if (OPAL_SUCCESS != (rc = ext2x_value_unload(oinfo, &data[n].value))) {
1187             OBJ_RELEASE(opalcaddy);
1188             return ext2x_convert_opalrc(rc);
1189         }
1190     }
1191 
1192     /* pass the call upwards */
1193     if (OPAL_SUCCESS != (rc = host_module->allocate(&requestor, odir,
1194                                                     &opalcaddy->info,
1195                                                     info_cbfunc, opalcaddy))) {
1196         OBJ_RELEASE(opalcaddy);
1197         return ext2x_convert_opalrc(rc);
1198     }
1199 
1200     return PMIX_SUCCESS;
1201 
1202 }
1203 
1204 static pmix_status_t server_job_control(const pmix_proc_t *proct,
1205                                         const pmix_proc_t targets[], size_t ntargets,
1206                                         const pmix_info_t directives[], size_t ndirs,
1207                                         pmix_info_cbfunc_t cbfunc, void *cbdata)
1208 {
1209     ext2x_opalcaddy_t *opalcaddy;
1210     opal_process_name_t requestor;
1211     int rc;
1212     size_t n;
1213     opal_value_t *oinfo;
1214     opal_namelist_t *nm;
1215 
1216     if (NULL == host_module || NULL == host_module->job_control) {
1217         return PMIX_ERR_NOT_SUPPORTED;
1218     }
1219 
1220     /* setup the caddy */
1221     opalcaddy = OBJ_NEW(ext2x_opalcaddy_t);
1222     opalcaddy->infocbfunc = cbfunc;
1223     opalcaddy->cbdata = cbdata;
1224 
1225     /* convert the requestor */
1226     if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&requestor.jobid, proct->nspace))) {
1227         OBJ_RELEASE(opalcaddy);
1228         return ext2x_convert_opalrc(rc);
1229     }
1230     requestor.vpid = ext2x_convert_rank(proct->rank);
1231 
1232     /* convert the targets */
1233     for (n=0; n < ntargets; n++) {
1234         nm = OBJ_NEW(opal_namelist_t);
1235         opal_list_append(&opalcaddy->procs, &nm->super);
1236         if (OPAL_SUCCESS != (rc = opal_convert_string_to_jobid(&nm->name.jobid, targets[n].nspace))) {
1237             OBJ_RELEASE(opalcaddy);
1238             return ext2x_convert_opalrc(rc);
1239         }
1240         nm->name.vpid = ext2x_convert_rank(targets[n].rank);
1241     }
1242 
1243     /* convert the directives */
1244     for (n=0; n < ndirs; n++) {
1245         oinfo = OBJ_NEW(opal_value_t);
1246         opal_list_append(&opalcaddy->info, &oinfo->super);
1247         oinfo->key = strdup(directives[n].key);
1248         if (OPAL_SUCCESS != (rc = ext2x_value_unload(oinfo, &directives[n].value))) {
1249             OBJ_RELEASE(opalcaddy);
1250             return ext2x_convert_opalrc(rc);
1251         }
1252     }
1253 
1254     /* pass the call upwards */
1255     if (OPAL_SUCCESS != (rc = host_module->job_control(&requestor,
1256                                                        &opalcaddy->procs,
1257                                                        &opalcaddy->info,
1258                                                        info_cbfunc, opalcaddy))) {
1259         OBJ_RELEASE(opalcaddy);
1260         return ext2x_convert_opalrc(rc);
1261     }
1262 
1263     return PMIX_SUCCESS;
1264 }
/* [<][>][^][v][top][bottom][index][help] */
root/opal/mca/pmix/ext2x/ext2x_server_north.c

DEFINITIONS
