ports/openmpi
ports/openmpi
1
1
Форкнуть 0
Код Релизы Активность

Add a new hierarchical collective grpcomm component that performs modex and barrier across the procs instead of the daemons. Modeled on the tuned collectives. Collective code is in grpcomm base for eventual use by the daemon-based components as well.

Просмотр исходного кода 
This commit was SVN r20337.
Этот коммит содержится в:
Ralph Castain 2009-01-23 21:57:51 +00:00
родитель 5eed7a0bec
Коммит c6c5bc17a0
9 изменённых файлов: 1556 добавлений и 0 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

1
orte/mca/grpcomm/base/Makefile.am
Просмотреть файл

@ -29,6 +29,7 @@ if !ORTE_DISABLE_FULL_SUPPORT
libmca_grpcomm_la_SOURCES += \
base/grpcomm_base_allgather.c \
base/grpcomm_base_modex.c \
base/grpcomm_base_coll.c \
base/grpcomm_base_receive.c
endif

8
orte/mca/grpcomm/base/base.h
Просмотреть файл

@ -82,6 +82,14 @@ ORTE_DECLSPEC int orte_grpcomm_base_load_modex_data(orte_process_name_t *proc,
ORTE_DECLSPEC int orte_grpcomm_base_comm_start(void);
ORTE_DECLSPEC int orte_grpcomm_base_comm_stop(void);
/* Tuned collectives */
ORTE_DECLSPEC void orte_grpcomm_base_coll_recv(int status, orte_process_name_t* sender,
opal_buffer_t* buffer, orte_rml_tag_t tag,
void* cbdata);
ORTE_DECLSPEC int orte_grpcomm_base_allgather(opal_buffer_t *sendbuf, opal_buffer_t *recvbuf, int32_t num_entries,
orte_jobid_t jobid, orte_vpid_t np, orte_vpid_t step);
#endif /* ORTE_DISABLE_FULL_SUPPORT */
END_C_DECLS

463
orte/mca/grpcomm/base/grpcomm_base_coll.c Обычный файл
Просмотреть файл

@ -0,0 +1,463 @@
/*
* Copyright (c) 2004-2007 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2007 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-2008 Los Alamos National Security, LLC.
* All rights reserved.
* $COPYRIGHT$
*
* Additional copyrights may follow
*
* $HEADER$
*
*/
#include "orte_config.h"
#include "orte/types.h"
#include "orte_config.h"
#include "orte/constants.h"
#include "orte/types.h"
#include <string.h>
#ifdef HAVE_SYS_TIME_H
#include <sys/time.h>
#endif /* HAVE_SYS_TIME_H */
#include "opal/threads/condition.h"
#include "orte/util/show_help.h"
#include "opal/util/bit_ops.h"
#include "opal/class/opal_hash_table.h"
#include "orte/util/proc_info.h"
#include "opal/dss/dss.h"
#include "orte/mca/errmgr/errmgr.h"
#include "orte/mca/odls/odls_types.h"
#include "orte/mca/rml/rml.h"
#include "orte/runtime/orte_globals.h"
#include "orte/util/name_fns.h"
#include "orte/orted/orted.h"
#include "orte/runtime/orte_wait.h"
#include "orte/mca/grpcomm/base/base.h"
/*************** TUNED COLLECTIVES FOR GRPCOMM MODULES **************/
/**** AVAILABLE ALGORITHMS ****/
static int twoproc(opal_buffer_t *sendbuf, opal_buffer_t *recvbuf, int32_t num_entries,
orte_jobid_t jobid, orte_vpid_t step);
static int bruck(opal_buffer_t *sendbuf, opal_buffer_t *recvbuf, int32_t num_entries,
orte_jobid_t jobid, orte_vpid_t np, orte_vpid_t step);
static int recursivedoubling(opal_buffer_t *sendbuf, opal_buffer_t *recvbuf, int32_t num_entries,
orte_jobid_t jobid, orte_vpid_t np, orte_vpid_t step);
/**** LOCAL VARIABLES USED IN COLLECTIVES ****/
static int num_recvd;
static opal_buffer_t bucket;
/* Receive and process collective messages */
static void process_coll_msg(int fd, short event, void *data)
{
orte_message_event_t *mev = (orte_message_event_t*)data;
/* transfer the data to the collecting bucket */
opal_dss.copy_payload(&bucket, mev->buffer);
/* cleanup */
OBJ_RELEASE(mev);
/* increment the number recvd */
num_recvd++;
}
void orte_grpcomm_base_coll_recv(int status, orte_process_name_t* sender,
opal_buffer_t* buffer, orte_rml_tag_t tag,
void* cbdata)
{
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:coll:receive got message from %s",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
ORTE_NAME_PRINT(sender)));
/* don't process this right away - we need to get out of the recv before
* we process the message as it may ask us to do something that involves
* more messaging! Instead, setup an event so that the message gets processed
* as soon as we leave the recv.
*
* The macro makes a copy of the buffer, which we release above - the incoming
* buffer, however, is NOT released here, although its payload IS transferred
* to the message buffer for later processing
*/
ORTE_MESSAGE_EVENT(sender, buffer, tag, process_coll_msg);
return;
}
/*
* Switchyard for selecting the collective algorithm to use
*/
int orte_grpcomm_base_allgather(opal_buffer_t *sendbuf, opal_buffer_t *recvbuf, int32_t num_entries,
orte_jobid_t jobid, orte_vpid_t np, orte_vpid_t step)
{
bool has_one;
orte_vpid_t n;
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:coll:allgather called with %d entries np %d step %d",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
num_entries, (int)np, (int)step));
/* if we only have one proc participating, just copy the data across and return */
if (1 == np) {
opal_dss.pack(recvbuf, &num_entries, 1, OPAL_INT32);
return opal_dss.copy_payload(recvbuf, sendbuf);
}
if (2 == np) {
/* only two procs in collective */
return twoproc(sendbuf, recvbuf, num_entries, jobid, step);
}
/* if we have power of 2 participants, use recursive doubling - otherwise,
* use bruck algorithm
*/
has_one = false;
n = np;
for ( ; n > 0; n >>= 1) {
if (n & 0x1) {
if (has_one) {
return bruck(sendbuf, recvbuf, num_entries, jobid, np, step);
}
has_one = true;
}
}
/* must be power of two! */
return recursivedoubling(sendbuf, recvbuf, num_entries, jobid, np, step);
}
/*
* The Two-Proc Algorithm
*
* One sends to zero, zero waits to recv from one
* Zero adds its data to message, sends result back to one
*/
static int twoproc(opal_buffer_t *sendbuf, opal_buffer_t *recvbuf, int32_t num_entries,
orte_jobid_t jobid, orte_vpid_t step)
{
orte_process_name_t peer;
int32_t num_remote, cnt;
int rc;
opal_buffer_t buf;
peer.jobid = jobid;
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:coll:two-proc algo employed",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME)));
if (0 == ORTE_PROC_MY_NAME->vpid) {
/* I send first */
peer.vpid = step;
/* setup a temp buffer so I can inform the other side as to the
* number of entries in my buffer
*/
OBJ_CONSTRUCT(&buf, opal_buffer_t);
opal_dss.pack(&buf, &num_entries, 1, OPAL_INT32);
opal_dss.copy_payload(&buf, sendbuf);
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:coll:two-proc sending to %s",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
ORTE_NAME_PRINT(&peer)));
if (0 > (rc = orte_rml.send_buffer(&peer, &buf, ORTE_RML_TAG_DAEMON_COLLECTIVE, 0))) {
ORTE_ERROR_LOG(rc);
return rc;
}
OBJ_DESTRUCT(&buf);
/* wait for reply */
num_recvd = 0;
OBJ_CONSTRUCT(&bucket, opal_buffer_t);
if (ORTE_SUCCESS != (rc = orte_rml.recv_buffer_nb(ORTE_NAME_WILDCARD,
ORTE_RML_TAG_DAEMON_COLLECTIVE,
ORTE_RML_NON_PERSISTENT,
orte_grpcomm_base_coll_recv,
NULL))) {
ORTE_ERROR_LOG(rc);
}
ORTE_PROGRESSED_WAIT(false, num_recvd, 1);
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:coll:two-proc got my return message",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME)));
} else {
/* if I am not the start, then I recv first */
num_recvd = 0;
OBJ_CONSTRUCT(&bucket, opal_buffer_t);
if (ORTE_SUCCESS != (rc = orte_rml.recv_buffer_nb(ORTE_NAME_WILDCARD,
ORTE_RML_TAG_DAEMON_COLLECTIVE,
ORTE_RML_NON_PERSISTENT,
orte_grpcomm_base_coll_recv,
NULL))) {
ORTE_ERROR_LOG(rc);
}
ORTE_PROGRESSED_WAIT(false, num_recvd, 1);
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:coll:two-proc got my starting message",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME)));
/* send my data back */
OBJ_CONSTRUCT(&buf, opal_buffer_t);
opal_dss.pack(&buf, &num_entries, 1, OPAL_INT32);
opal_dss.copy_payload(&buf, sendbuf);
peer.vpid = 0;
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:coll:two-proc sending to %s",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
ORTE_NAME_PRINT(&peer)));
if (0 > (rc = orte_rml.send_buffer(&peer, &buf, ORTE_RML_TAG_DAEMON_COLLECTIVE, 0))) {
ORTE_ERROR_LOG(rc);
return rc;
}
OBJ_DESTRUCT(&buf);
}
/* extract the number of entries in the remote buffer */
cnt = 1;
if (ORTE_SUCCESS != (rc = opal_dss.unpack(&bucket, &num_remote, &cnt, OPAL_INT32))) {
ORTE_ERROR_LOG(rc);
return rc;
}
/* output of a collective begins with the total number of entries */
num_remote += num_entries;
if (ORTE_SUCCESS != (rc = opal_dss.pack(recvbuf, &num_remote, 1, OPAL_INT32))) {
ORTE_ERROR_LOG(rc);
return rc;
}
/* xfer my data */
opal_dss.copy_payload(recvbuf, sendbuf);
/* xfer the recvd data */
opal_dss.copy_payload(recvbuf, &bucket);
/* cleanup */
OBJ_DESTRUCT(&bucket);
return ORTE_SUCCESS;
}
/* For a complete description of this algorithm, please look at
* ompi/mca/coll/tuned/coll_tuned_allgather.c
*/
static int bruck(opal_buffer_t *sendbuf, opal_buffer_t *recvbuf, int32_t num_entries,
orte_jobid_t jobid, orte_vpid_t np, orte_vpid_t step)
{
orte_vpid_t rank, distance, stp;
orte_process_name_t peer;
int32_t num_remote, total_entries, cnt;
opal_buffer_t collection, buf;
int rc;
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:coll:bruck algo employed",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME)));
/* initialize */
total_entries = num_entries;
/* start by seeding the collection with our own data */
OBJ_CONSTRUCT(&collection, opal_buffer_t);
opal_dss.copy_payload(&collection, sendbuf);
/* collective is constrained to take place within the specified jobid */
peer.jobid = jobid;
/* Communication step:
At every step i, rank r:
- doubles the distance
- sends message containing all data collected so far to rank r - distance
- receives message containing all data collected so far from rank (r + distance)
*/
/* find my position in the group of participants - it always starts at rank=0. This
* value is the "rank" we will use in the algo
*/
rank = (ORTE_PROC_MY_NAME->vpid) / step;
for (distance = 1; distance < np; distance <<= 1) {
/* first send my current contents */
stp = (rank - distance + np) % np;
peer.vpid = (stp * step);
OBJ_CONSTRUCT(&buf, opal_buffer_t);
opal_dss.pack(&buf, &total_entries, 1, OPAL_INT32);
opal_dss.copy_payload(&buf, &collection);
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:coll:bruck sending to %s",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
ORTE_NAME_PRINT(&peer)));
if (0 > (rc = orte_rml.send_buffer(&peer, &buf, ORTE_RML_TAG_DAEMON_COLLECTIVE, 0))) {
ORTE_ERROR_LOG(rc);
return rc;
}
OBJ_DESTRUCT(&buf);
/* now setup to recv from my other partner */
num_recvd = 0;
stp = (rank + distance) % np;
peer.vpid = (stp * step);
OBJ_CONSTRUCT(&bucket, opal_buffer_t);
if (ORTE_SUCCESS != (rc = orte_rml.recv_buffer_nb(&peer,
ORTE_RML_TAG_DAEMON_COLLECTIVE,
ORTE_RML_NON_PERSISTENT,
orte_grpcomm_base_coll_recv,
NULL))) {
ORTE_ERROR_LOG(rc);
return rc;
}
/* and wait for it to get here */
ORTE_PROGRESSED_WAIT(false, num_recvd, 1);
/* extract the number of entries in the remote buffer */
cnt = 1;
if (ORTE_SUCCESS != (rc = opal_dss.unpack(&bucket, &num_remote, &cnt, OPAL_INT32))) {
ORTE_ERROR_LOG(rc);
return rc;
}
/* add it to our running total */
total_entries += num_remote;
/* transfer the data to our collection */
opal_dss.copy_payload(&collection, &bucket);
/* cleanup */
OBJ_DESTRUCT(&bucket);
}
/* output of a collective begins with the total number of entries */
if (ORTE_SUCCESS != (rc = opal_dss.pack(recvbuf, &total_entries, 1, OPAL_INT32))) {
ORTE_ERROR_LOG(rc);
return rc;
}
/* transfer the collected data */
opal_dss.copy_payload(recvbuf, &collection);
/* cleanup */
OBJ_DESTRUCT(&collection);
return ORTE_SUCCESS;
}
/* For a complete description of this algorithm, please look at
* ompi/mca/coll/tuned/coll_tuned_allgather.c
*/
static int recursivedoubling(opal_buffer_t *sendbuf, opal_buffer_t *recvbuf, int32_t num_entries,
orte_jobid_t jobid, orte_vpid_t np, orte_vpid_t step)
{
orte_vpid_t rank, distance, stp;
int32_t num_remote, total_entries, cnt;
opal_buffer_t collection, buf;
orte_process_name_t peer;
int rc;
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:coll:recdub algo employed",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME)));
/* initialize */
total_entries = num_entries;
/* start by seeding the collection with our own data */
OBJ_CONSTRUCT(&collection, opal_buffer_t);
opal_dss.copy_payload(&collection, sendbuf);
/* collective is constrained to take place within the specified jobid */
peer.jobid = jobid;
/* Communication step:
At every step i, rank r:
- exchanges message containing all data collected so far with rank peer = (r ^ 2^i).
*/
/* find my position in the group of participants - it always starts at rank=0. This
* value is the "rank" we will use in the algo
*/
rank = (ORTE_PROC_MY_NAME->vpid) / step;
for (distance = 0x1; distance < np; distance<<=1) {
/* first send my current contents */
stp = rank ^ distance;
peer.vpid = (stp * step);
OBJ_CONSTRUCT(&buf, opal_buffer_t);
opal_dss.pack(&buf, &total_entries, 1, OPAL_INT32);
opal_dss.copy_payload(&buf, &collection);
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:coll:recdub sending to %s",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
ORTE_NAME_PRINT(&peer)));
if (0 > (rc = orte_rml.send_buffer(&peer, &buf, ORTE_RML_TAG_DAEMON_COLLECTIVE, 0))) {
ORTE_ERROR_LOG(rc);
return rc;
}
OBJ_DESTRUCT(&buf);
/* now setup to recv from my other partner */
num_recvd = 0;
OBJ_CONSTRUCT(&bucket, opal_buffer_t);
if (ORTE_SUCCESS != (rc = orte_rml.recv_buffer_nb(&peer,
ORTE_RML_TAG_DAEMON_COLLECTIVE,
ORTE_RML_NON_PERSISTENT,
orte_grpcomm_base_coll_recv,
NULL))) {
ORTE_ERROR_LOG(rc);
return rc;
}
/* and wait for it to get here */
ORTE_PROGRESSED_WAIT(false, num_recvd, 1);
/* extract the number of entries in the remote buffer */
cnt = 1;
if (ORTE_SUCCESS != (rc = opal_dss.unpack(&bucket, &num_remote, &cnt, OPAL_INT32))) {
ORTE_ERROR_LOG(rc);
return rc;
}
/* add it to our running total */
total_entries += num_remote;
/* transfer the data to our collection */
opal_dss.copy_payload(&collection, &bucket);
/* cleanup */
OBJ_DESTRUCT(&bucket);
}
/* output of a collective begins with the total number of entries */
if (ORTE_SUCCESS != (rc = opal_dss.pack(recvbuf, &total_entries, 1, OPAL_INT32))) {
ORTE_ERROR_LOG(rc);
return rc;
}
/* transfer the collected data */
opal_dss.copy_payload(recvbuf, &collection);
/* cleanup */
OBJ_DESTRUCT(&collection);
return ORTE_SUCCESS;
}

43
orte/mca/grpcomm/hier/Makefile.am Обычный файл
Просмотреть файл

@ -0,0 +1,43 @@
#
# Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
# University Research and Technology
# Corporation. All rights reserved.
# Copyright (c) 2004-2005 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$
#
# Additional copyrights may follow
#
# $HEADER$
#
sources = \
grpcomm_hier.h \
grpcomm_hier_module.c \
grpcomm_hier_component.c
# Make the output library in this directory, and name it either
# mca_<type>_<name>.la (for DSO builds) or libmca_<type>_<name>.la
# (for static builds).
if OMPI_BUILD_grpcomm_hier_DSO
component_noinst =
component_install = mca_grpcomm_hier.la
else
component_noinst = libmca_grpcomm_hier.la
component_install =
endif
mcacomponentdir = $(pkglibdir)
mcacomponent_LTLIBRARIES = $(component_install)
mca_grpcomm_hier_la_SOURCES = $(sources)
mca_grpcomm_hier_la_LDFLAGS = -module -avoid-version
noinst_LTLIBRARIES = $(component_noinst)
libmca_grpcomm_hier_la_SOURCES =$(sources)
libmca_grpcomm_hier_la_LDFLAGS = -module -avoid-version

13
orte/mca/grpcomm/hier/configure.m4 Обычный файл
Просмотреть файл

@ -0,0 +1,13 @@
# -*- shell-script -*-
#
# Copyright (c) 2007 Sandia National Laboratories. All rights reserved.
# $COPYRIGHT$
#
# Additional copyrights may follow
#
# $HEADER$
#
# MCA_grpcomm_hier_CONFIG([action-if-found], [action-if-not-found])
# -----------------------------------------------------------
AC_DEFUN([MCA_grpcomm_hier_CONFIG], [$1])

30
orte/mca/grpcomm/hier/configure.params Обычный файл
Просмотреть файл

@ -0,0 +1,30 @@
# -*- shell-script -*-
#
# Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
# University Research and Technology
# Corporation. All rights reserved.
# Copyright (c) 2004-2005 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) 2007 Los Alamos National Security, LLC. All rights
# reserved.
# $COPYRIGHT$
#
# Additional copyrights may follow
#
# $HEADER$
#
# Specific to this module
PARAM_CONFIG_FILES="Makefile"
#
# Set the config priority so that this
# component will build for all environs -except-
# those special ones that do not support it
PARAM_CONFIG_PRIORITY=10

57
orte/mca/grpcomm/hier/grpcomm_hier.h Обычный файл
Просмотреть файл

@ -0,0 +1,57 @@
/* -*- C -*-
*
* Copyright (c) 2004-2008 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2006 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$
*
* Additional copyrights may follow
*
* $HEADER$
*
*/
#ifndef GRPCOMM_HIER_H
#define GRPCOMM_HIER_H
#include "orte_config.h"
#include "orte/types.h"
#include "opal/threads/mutex.h"
#include "opal/threads/condition.h"
#include "opal/class/opal_object.h"
#include "orte/mca/grpcomm/grpcomm.h"
BEGIN_C_DECLS
/*
* Component open / close
*/
int orte_grpcomm_hier_open(void);
int orte_grpcomm_hier_close(void);
int orte_grpcomm_hier_component_query(mca_base_module_t **module, int *priority);
/* Hier component */
typedef struct {
orte_grpcomm_base_component_t super;
orte_vpid_t num_nodes;
orte_vpid_t step;
} orte_grpcomm_hier_component_t;
/*
* Grpcomm interfaces
*/
ORTE_MODULE_DECLSPEC extern orte_grpcomm_hier_component_t mca_grpcomm_hier_component;
extern orte_grpcomm_base_module_t orte_grpcomm_hier_module;
END_C_DECLS
#endif

104
orte/mca/grpcomm/hier/grpcomm_hier_component.c Обычный файл
Просмотреть файл

@ -0,0 +1,104 @@
/* -*- C -*-
*
* Copyright (c) 2004-2008 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2005 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$
*
* Additional copyrights may follow
*
* $HEADER$
*/
/** @file:
*
*/
/*
* includes
*/
#include "orte_config.h"
#include "orte/constants.h"
#include "opal/threads/mutex.h"
#include "opal/class/opal_list.h"
#include "orte/util/show_help.h"
#include "opal/mca/mca.h"
#include "opal/mca/base/mca_base_param.h"
#include "orte/util/proc_info.h"
#include "orte/mca/errmgr/errmgr.h"
#include "orte/mca/rml/rml.h"
#include "grpcomm_hier.h"
/*
* Struct of function pointers that need to be initialized
*/
orte_grpcomm_hier_component_t mca_grpcomm_hier_component = {
{
{
ORTE_GRPCOMM_BASE_VERSION_2_0_0,
"hier", /* MCA module name */
ORTE_MAJOR_VERSION, /* MCA module major version */
ORTE_MINOR_VERSION, /* MCA module minor version */
ORTE_RELEASE_VERSION, /* MCA module release version */
orte_grpcomm_hier_open, /* module open */
orte_grpcomm_hier_close, /* module close */
orte_grpcomm_hier_component_query /* module query */
},
{
/* The component is checkpoint ready */
MCA_BASE_METADATA_PARAM_CHECKPOINT
}
}
};
/* Open the component */
int orte_grpcomm_hier_open(void)
{
return ORTE_SUCCESS;
}
int orte_grpcomm_hier_close(void)
{
return ORTE_SUCCESS;
}
int orte_grpcomm_hier_component_query(mca_base_module_t **module, int *priority)
{
mca_base_component_t *c = &mca_grpcomm_hier_component.super.base_version;
int tmp;
/* check for required params */
mca_base_param_reg_int(c, "num_nodes",
"How many nodes are in the job (must be > 0)",
false, false, -1, &tmp);
if (tmp < 0) {
*module = NULL;
return ORTE_ERROR;
}
mca_grpcomm_hier_component.num_nodes = tmp;
mca_base_param_reg_int(c, "step",
"Step in local_rank=0 vpids between nodes (must be > 0)",
false, false, -1, &tmp);
if (tmp < 0) {
*module = NULL;
return ORTE_ERROR;
}
mca_grpcomm_hier_component.step = tmp;
/* we need to be selected */
*priority = 100;
*module = (mca_base_module_t *)&orte_grpcomm_hier_module;
return ORTE_SUCCESS;
}

837
orte/mca/grpcomm/hier/grpcomm_hier_module.c Обычный файл
Просмотреть файл

@ -0,0 +1,837 @@
/*
* Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2005 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) 2007 Sun Microsystems, Inc. All rights reserved.
* $COPYRIGHT$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#include "orte_config.h"
#include "orte/constants.h"
#include "orte/types.h"
#include <string.h>
#include <fcntl.h>
#include "opal/threads/condition.h"
#include "opal/util/bit_ops.h"
#include "opal/class/opal_hash_table.h"
#include "opal/dss/dss.h"
#include "opal/runtime/opal.h"
#include "orte/mca/errmgr/errmgr.h"
#include "orte/mca/ess/ess.h"
#include "orte/mca/odls/odls_types.h"
#include "orte/mca/rml/rml.h"
#include "orte/util/name_fns.h"
#include "orte/util/show_help.h"
#include "orte/util/proc_info.h"
#include "orte/util/nidmap.h"
#include "orte/orted/orted.h"
#include "orte/runtime/orte_wait.h"
#include "orte/runtime/orte_globals.h"
#include "orte/mca/grpcomm/base/base.h"
#include "grpcomm_hier.h"
/* Static API's */
static int init(void);
static void finalize(void);
static int xcast(orte_jobid_t job,
opal_buffer_t *buffer,
orte_rml_tag_t tag);
static int allgather(opal_buffer_t *sbuf, opal_buffer_t *rbuf);
static int barrier(void);
static int modex(opal_list_t *procs);
static int set_proc_attr(const char *attr_name, const void *data, size_t size);
static int get_proc_attr(const orte_process_name_t proc,
const char * attribute_name, void **val,
size_t *size);
/* Module def */
orte_grpcomm_base_module_t orte_grpcomm_hier_module = {
init,
finalize,
xcast,
allgather,
orte_grpcomm_base_allgather_list,
barrier,
set_proc_attr,
get_proc_attr,
modex,
orte_grpcomm_base_purge_proc_attrs
};
/* Local data */
static orte_local_rank_t my_local_rank;
static opal_list_t my_local_peers;
static orte_process_name_t my_local_rank_zero_proc;
static int num_local_peers;
static bool coll_initialized = false;
/**
* Initialize the module
*/
static int init(void)
{
int rc;
OBJ_CONSTRUCT(&my_local_peers, opal_list_t);
if (ORTE_SUCCESS != (rc = orte_grpcomm_base_modex_init())) {
ORTE_ERROR_LOG(rc);
return rc;
}
return rc;
}
/**
* Finalize the module
*/
static void finalize(void)
{
opal_list_item_t *item;
orte_grpcomm_base_modex_finalize();
while (NULL != (item = opal_list_remove_first(&my_local_peers))) {
OBJ_RELEASE(item);
}
OBJ_DESTRUCT(&my_local_peers);
}
/**
* A "broadcast-like" function to a job's processes.
* @param jobid The job whose processes are to receive the message
* @param buffer The data to broadcast
*/
static int xcast(orte_jobid_t job,
opal_buffer_t *buffer,
orte_rml_tag_t tag)
{
int rc = ORTE_SUCCESS;
opal_buffer_t buf;
orte_daemon_cmd_flag_t command;
OPAL_OUTPUT_VERBOSE((1, orte_grpcomm_base_output,
"%s grpcomm:xcast sent to job %s tag %ld",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
ORTE_JOBID_PRINT(job), (long)tag));
/* if there is no message to send, then just return ok */
if (NULL == buffer) {
return ORTE_SUCCESS;
}
/* setup a buffer to handle the xcast command */
OBJ_CONSTRUCT(&buf, opal_buffer_t);
/* all we need to do is send this to the HNP - the relay logic
* will ensure everyone else gets it! So tell the HNP to
* process and relay it. The HNP will use the routed.get_routing_tree
* to find out who it should relay the message to.
*/
command = ORTE_DAEMON_PROCESS_AND_RELAY_CMD;
if (ORTE_SUCCESS != (rc = opal_dss.pack(&buf, &command, 1, ORTE_DAEMON_CMD))) {
ORTE_ERROR_LOG(rc);
goto CLEANUP;
}
/* pack the target jobid and tag for use in relay */
if (ORTE_SUCCESS != (rc = opal_dss.pack(&buf, &job, 1, ORTE_JOBID))) {
ORTE_ERROR_LOG(rc);
goto CLEANUP;
}
if (ORTE_SUCCESS != (rc = opal_dss.pack(&buf, &tag, 1, ORTE_RML_TAG))) {
ORTE_ERROR_LOG(rc);
goto CLEANUP;
}
/* if this isn't intended for the daemon command tag, then we better
* tell the daemon to deliver it to the procs, and what job is supposed
* to get it - this occurs when a caller just wants to send something
* to all the procs in a job. In that use-case, the caller doesn't know
* anything about inserting daemon commands or what routing algo might
* be used, so we have to help them out a little. Functions that are
* sending commands to the daemons themselves are smart enough to know
* what they need to do.
*/
if (ORTE_RML_TAG_DAEMON != tag) {
command = ORTE_DAEMON_MESSAGE_LOCAL_PROCS;
if (ORTE_SUCCESS != (rc = opal_dss.pack(&buf, &command, 1, ORTE_DAEMON_CMD))) {
ORTE_ERROR_LOG(rc);
goto CLEANUP;
}
if (ORTE_SUCCESS != (rc = opal_dss.pack(&buf, &job, 1, ORTE_JOBID))) {
ORTE_ERROR_LOG(rc);
goto CLEANUP;
}
if (ORTE_SUCCESS != (rc = opal_dss.pack(&buf, &tag, 1, ORTE_RML_TAG))) {
ORTE_ERROR_LOG(rc);
goto CLEANUP;
}
}
/* copy the payload into the new buffer - this is non-destructive, so our
* caller is still responsible for releasing any memory in the buffer they
* gave to us
*/
if (ORTE_SUCCESS != (rc = opal_dss.copy_payload(&buf, buffer))) {
ORTE_ERROR_LOG(rc);
goto CLEANUP;
}
/* if I am the HNP, just set things up so the cmd processor gets called.
* We don't want to message ourselves as this can create circular logic
* in the RML. Instead, this macro will set a zero-time event which will
* cause the buffer to be processed by the cmd processor - probably will
* fire right away, but that's okay
* The macro makes a copy of the buffer, so it's okay to release it here
*/
if (orte_process_info.hnp) {
ORTE_MESSAGE_EVENT(ORTE_PROC_MY_NAME, &buf, ORTE_RML_TAG_DAEMON, orte_daemon_cmd_processor);
} else {
/* otherwise, send it to the HNP for relay */
if (0 > (rc = orte_rml.send_buffer(ORTE_PROC_MY_HNP, &buf, ORTE_RML_TAG_DAEMON, 0))) {
ORTE_ERROR_LOG(rc);
goto CLEANUP;
}
rc = ORTE_SUCCESS;
}
CLEANUP:
OBJ_DESTRUCT(&buf);
return rc;
}
/* the barrier is executed as an allgather with data length of zero */
static int barrier(void)
{
opal_buffer_t buf1, buf2;
int rc;
OPAL_OUTPUT_VERBOSE((1, orte_grpcomm_base_output,
"%s grpcomm:hier entering barrier",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME)));
OBJ_CONSTRUCT(&buf1, opal_buffer_t);
OBJ_CONSTRUCT(&buf2, opal_buffer_t);
if (ORTE_SUCCESS != (rc = allgather(&buf1, &buf2))) {
ORTE_ERROR_LOG(rc);
}
OBJ_DESTRUCT(&buf1);
OBJ_DESTRUCT(&buf2);
OPAL_OUTPUT_VERBOSE((1, orte_grpcomm_base_output,
"%s grpcomm:hier barrier complete",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME)));
return rc;
}
static opal_buffer_t allgather_buf;
static int allgather_num_recvd;
static void process_msg(int fd, short event, void *data)
{
int rc;
orte_message_event_t *mev = (orte_message_event_t*)data;
/* xfer the data */
if (ORTE_SUCCESS != (rc = opal_dss.copy_payload(&allgather_buf, mev->buffer))) {
ORTE_ERROR_LOG(rc);
}
allgather_num_recvd++;
/* release the message */
OBJ_RELEASE(mev);
}
static void allgather_recv(int status, orte_process_name_t* sender,
opal_buffer_t *buffer,
orte_rml_tag_t tag, void *cbdata)
{
int rc;
/* don't process this right away - we need to get out of the recv before
* we process the message as it may ask us to do something that involves
* more messaging! Instead, setup an event so that the message gets processed
* as soon as we leave the recv.
*
* The macro makes a copy of the buffer, which we release above - the incoming
* buffer, however, is NOT released here, although its payload IS transferred
* to the message buffer for later processing
*/
ORTE_MESSAGE_EVENT(sender, buffer, tag, process_msg);
/* reissue the recv */
if (ORTE_SUCCESS != (rc = orte_rml.recv_buffer_nb(ORTE_NAME_WILDCARD, ORTE_RML_TAG_ALLGATHER,
ORTE_RML_NON_PERSISTENT, allgather_recv, NULL))) {
ORTE_ERROR_LOG(rc);
}
}
static int allgather(opal_buffer_t *sbuf, opal_buffer_t *rbuf)
{
int rc=ORTE_SUCCESS;
opal_list_item_t *item;
orte_namelist_t *nm;
opal_buffer_t final_buf;
OPAL_OUTPUT_VERBOSE((1, orte_grpcomm_base_output,
"%s grpcomm:hier entering allgather",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME)));
/* have I initialized my local info? */
if (!coll_initialized) {
orte_process_name_t proc;
orte_vpid_t v, vmax;
/* no - cycle through the procs to find those that are local */
proc.jobid = ORTE_PROC_MY_NAME->jobid;
vmax = ORTE_VPID_MAX;
my_local_rank = 0;
num_local_peers = 0; /* don't count myself */
for (v=0; v < orte_process_info.num_procs; v++) {
/* ignore if this is me */
if (v == ORTE_PROC_MY_NAME->vpid) {
continue;
}
proc.vpid = v;
if (!orte_ess.proc_is_local(&proc)) {
continue;
}
/* add this proc to our list of local peers */
nm = OBJ_NEW(orte_namelist_t);
nm->name.jobid = proc.jobid;
nm->name.vpid = proc.vpid;
opal_list_append(&my_local_peers, &nm->item);
/* keep count */
num_local_peers++;
/* is this our locally lowest rank? */
if (v < vmax) {
vmax = v;
}
/* is this rank lower than mine? */
if (v < ORTE_PROC_MY_NAME->vpid) {
my_local_rank++;
}
}
/* if I am not the local_rank=0 proc, record who is */
if (0 != my_local_rank) {
my_local_rank_zero_proc.jobid = ORTE_PROC_MY_NAME->jobid;
my_local_rank_zero_proc.vpid = vmax;
}
/* flag that I have initialized things */
coll_initialized = true;
}
/* if I am not local rank = 0 */
if (0 != my_local_rank) {
/* send our data to the local_rank=0 proc on this node */
if (0 > (rc = orte_rml.send_buffer(&my_local_rank_zero_proc, sbuf, ORTE_RML_TAG_ALLGATHER, 0))) {
ORTE_ERROR_LOG(rc);
return rc;
}
/* setup to get return buffer */
OBJ_CONSTRUCT(&allgather_buf, opal_buffer_t);
/* now receive the final result. Be sure to do this in
* a manner that allows us to return without being in a recv!
*/
allgather_num_recvd = 0;
rc = orte_rml.recv_buffer_nb(ORTE_NAME_WILDCARD, ORTE_RML_TAG_ALLGATHER,
ORTE_RML_NON_PERSISTENT, allgather_recv, NULL);
if (rc != ORTE_SUCCESS) {
ORTE_ERROR_LOG(rc);
return rc;
}
ORTE_PROGRESSED_WAIT(false, allgather_num_recvd, 1);
/* copy payload to the caller's buffer */
if (ORTE_SUCCESS != (rc = opal_dss.copy_payload(rbuf, &allgather_buf))) {
ORTE_ERROR_LOG(rc);
}
OBJ_DESTRUCT(&allgather_buf);
} else {
/* I am local_rank = 0 on this node! */
/* setup to recv data from the procs that share this node with me */
OBJ_CONSTRUCT(&allgather_buf, opal_buffer_t);
/* seed it with my own data */
opal_dss.copy_payload(&allgather_buf, sbuf);
/* wait to receive their data. Be sure to do this in
* a manner that allows us to return without being in a recv!
*/
allgather_num_recvd = 0;
rc = orte_rml.recv_buffer_nb(ORTE_NAME_WILDCARD, ORTE_RML_TAG_ALLGATHER,
ORTE_RML_NON_PERSISTENT, allgather_recv, NULL);
if (rc != ORTE_SUCCESS) {
ORTE_ERROR_LOG(rc);
return rc;
}
ORTE_PROGRESSED_WAIT(false, allgather_num_recvd, num_local_peers);
/* take the recv'd data and use one of the base collectives
* to exchange it with all other local_rank=0 procs in a scalable
* manner - the exact collective will depend upon the number of
* nodes in the job
*/
OBJ_CONSTRUCT(&final_buf, opal_buffer_t);
if (ORTE_SUCCESS != (rc = orte_grpcomm_base_allgather(&allgather_buf, rbuf, num_local_peers + 1,
ORTE_PROC_MY_NAME->jobid,
mca_grpcomm_hier_component.num_nodes,
mca_grpcomm_hier_component.step))) {
ORTE_ERROR_LOG(rc);
OBJ_DESTRUCT(&allgather_buf);
OBJ_DESTRUCT(&final_buf);
return rc;
}
OBJ_DESTRUCT(&allgather_buf); /* done with this */
/* distribute the results to our local peers */
for (item = opal_list_get_first(&my_local_peers);
item != opal_list_get_end(&my_local_peers);
item = opal_list_get_next(item)) {
nm = (orte_namelist_t*)item;
if (0 > (rc = orte_rml.send_buffer(&nm->name, rbuf, ORTE_RML_TAG_ALLGATHER, 0))) {
ORTE_ERROR_LOG(rc);
return rc;
}
}
}
OPAL_OUTPUT_VERBOSE((1, orte_grpcomm_base_output,
"%s grpcomm:hier allgather completed",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME)));
return ORTE_SUCCESS;
}
/*** MODEX SECTION ***/
static int do_modex(opal_list_t *procs)
{
opal_buffer_t buf, rbuf;
int32_t i, num_procs;
orte_std_cntr_t cnt, j, num_recvd_entries;
orte_process_name_t proc_name;
int rc=ORTE_SUCCESS;
int32_t arch;
bool modex_reqd;
orte_nid_t *nid;
OPAL_OUTPUT_VERBOSE((1, orte_grpcomm_base_output,
"%s grpcomm:hier:modex: performing modex",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME)));
/* setup the buffer that will actually be sent */
OBJ_CONSTRUCT(&buf, opal_buffer_t);
OBJ_CONSTRUCT(&rbuf, opal_buffer_t);
/* put our process name in the buffer so it can be unpacked later */
if (ORTE_SUCCESS != (rc = opal_dss.pack(&buf, ORTE_PROC_MY_NAME, 1, ORTE_NAME))) {
ORTE_ERROR_LOG(rc);
goto cleanup;
}
if (ORTE_SUCCESS != (rc = opal_dss.pack(&buf, &orte_process_info.arch, 1, OPAL_UINT32))) {
ORTE_ERROR_LOG(rc);
goto cleanup;
}
/* pack the entries we have received */
if (ORTE_SUCCESS != (rc = orte_grpcomm_base_pack_modex_entries(&buf, &modex_reqd))) {
ORTE_ERROR_LOG(rc);
goto cleanup;
}
OPAL_OUTPUT_VERBOSE((2, orte_grpcomm_base_output,
"%s grpcomm:hier:modex: executing allgather",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME)));
/* exchange the buffer with the list of peers (if provided) or all my peers */
if (NULL == procs) {
if (ORTE_SUCCESS != (rc = allgather(&buf, &rbuf))) {
ORTE_ERROR_LOG(rc);
goto cleanup;
}
} else {
if (ORTE_SUCCESS != (rc = orte_grpcomm_base_allgather_list(procs, &buf, &rbuf))) {
ORTE_ERROR_LOG(rc);
goto cleanup;
}
}
OPAL_OUTPUT_VERBOSE((2, orte_grpcomm_base_output,
"%s grpcomm:hier:modex: processing modex info",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME)));
/* process the results */
/* extract the number of procs that put data in the buffer */
cnt=1;
if (ORTE_SUCCESS != (rc = opal_dss.unpack(&rbuf, &num_procs, &cnt, OPAL_INT32))) {
ORTE_ERROR_LOG(rc);
goto cleanup;
}
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:hier:modex: received %ld data bytes from %d procs",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
(long)(rbuf.pack_ptr - rbuf.unpack_ptr), num_procs));
/* if the buffer doesn't have any more data, ignore it */
if (0 >= (rbuf.pack_ptr - rbuf.unpack_ptr)) {
goto cleanup;
}
/* otherwise, process it */
for (i=0; i < num_procs; i++) {
/* unpack the process name */
cnt=1;
if (ORTE_SUCCESS != (rc = opal_dss.unpack(&rbuf, &proc_name, &cnt, ORTE_NAME))) {
ORTE_ERROR_LOG(rc);
goto cleanup;
}
/* unpack its architecture */
cnt=1;
if (ORTE_SUCCESS != (rc = opal_dss.unpack(&rbuf, &arch, &cnt, OPAL_UINT32))) {
ORTE_ERROR_LOG(rc);
goto cleanup;
}
/* update the arch in the ESS
* RHC: DO NOT UPDATE ARCH IF THE PROC IS NOT IN OUR JOB. THIS IS A TEMPORARY
* FIX TO COMPENSATE FOR A PROBLEM IN THE CONNECT/ACCEPT CODE WHERE WE EXCHANGE
* INFO INCLUDING THE ARCH, BUT THEN DO A MODEX THAT ALSO INCLUDES THE ARCH. WE
* CANNOT UPDATE THE ARCH FOR JOBS OUTSIDE OUR OWN AS THE ESS HAS NO INFO ON
* THOSE PROCS/NODES - AND DOESN'T NEED IT AS THE MPI LAYER HAS ALREADY SET
* ITSELF UP AND DOES NOT NEED ESS SUPPORT FOR PROCS IN THE OTHER JOB
*
* EVENTUALLY, WE WILL SUPPORT THE ESS HAVING INFO ON OTHER JOBS FOR
* FAULT TOLERANCE PURPOSES - BUT NOT RIGHT NOW
*/
if (proc_name.jobid == ORTE_PROC_MY_NAME->jobid) {
if (ORTE_SUCCESS != (rc = orte_ess.update_arch(&proc_name, arch))) {
ORTE_ERROR_LOG(rc);
goto cleanup;
}
}
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:hier:modex: adding modex entry for proc %s",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
ORTE_NAME_PRINT(&proc_name)));
/* unpack the number of entries for this proc */
cnt=1;
if (ORTE_SUCCESS != (rc = opal_dss.unpack(&rbuf, &num_recvd_entries, &cnt, ORTE_STD_CNTR))) {
ORTE_ERROR_LOG(rc);
goto cleanup;
}
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:hier:modex adding %d entries for proc %s",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME), num_recvd_entries,
ORTE_NAME_PRINT(&proc_name)));
/* find this proc's node in the nidmap */
if (NULL == (nid = orte_util_lookup_nid(&proc_name))) {
/* proc wasn't found - return error */
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:hier:modex no nidmap entry for proc %s",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
ORTE_NAME_PRINT(&proc_name)));
ORTE_ERROR_LOG(ORTE_ERR_NOT_FOUND);
rc = ORTE_ERR_NOT_FOUND;
goto cleanup;
}
/*
* Extract the attribute names and values
*/
for (j = 0; j < num_recvd_entries; j++) {
size_t num_bytes;
orte_attr_t *attr;
attr = OBJ_NEW(orte_attr_t);
cnt = 1;
if (ORTE_SUCCESS != (rc = opal_dss.unpack(&rbuf, &(attr->name), &cnt, OPAL_STRING))) {
ORTE_ERROR_LOG(rc);
goto cleanup;
}
cnt = 1;
if (ORTE_SUCCESS != (rc = opal_dss.unpack(&rbuf, &num_bytes, &cnt, OPAL_SIZE))) {
ORTE_ERROR_LOG(rc);
goto cleanup;
}
attr->size = num_bytes;
if (num_bytes != 0) {
if (NULL == (attr->bytes = malloc(num_bytes))) {
ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
rc = ORTE_ERR_OUT_OF_RESOURCE;
goto cleanup;
}
cnt = (orte_std_cntr_t) num_bytes;
if (ORTE_SUCCESS != (rc = opal_dss.unpack(&rbuf, attr->bytes, &cnt, OPAL_BYTE))) {
ORTE_ERROR_LOG(rc);
goto cleanup;
}
}
/* add this to the node's attribute list */
opal_list_append(&nid->attrs, &attr->super);
}
}
cleanup:
OBJ_DESTRUCT(&buf);
OBJ_DESTRUCT(&rbuf);
return rc;
}
static int modex(opal_list_t *procs)
{
int rc=ORTE_SUCCESS;
int fd;
opal_byte_object_t bo, *boptr;
int32_t i, n;
char *nodename, *attr;
orte_nid_t **nd, *ndptr;
orte_attr_t *attrdata;
opal_buffer_t bobuf;
OPAL_OUTPUT_VERBOSE((1, orte_grpcomm_base_output,
"%s grpcomm:hier: modex entered",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME)));
/* if we were given a list of procs to modex with, then this is happening
* as part of a connect/accept operation. In this case, we -must- do the
* modex for two reasons:
*
* (a) the modex could involve procs from different mpiruns. In this case,
* there is no way for the two sets of procs to know which node the
* other procs are on, so we cannot use the profile_file to determine
* their contact info
*
* (b) in a comm_spawn, the parent job does not have a pidmap for the
* child job. Thus, it cannot know where the child procs are located,
* and cannot use the profile_file to determine their contact info
*
* We also do the modex if we are doing an opal_profile so that the
* HNP can collect our modex info.
*/
if (NULL != procs || opal_profile) {
if (ORTE_SUCCESS != (rc = do_modex(procs))) {
ORTE_ERROR_LOG(rc);
}
return rc;
} else if (OMPI_ENABLE_HETEROGENEOUS_SUPPORT) {
/* decide if we need to add the architecture to the modex. Check
* first to see if hetero is enabled - if not, then we clearly
* don't need to exchange arch's as they are all identical
*/
/* Case 1: If different apps in this job were built differently - e.g., some
* are built 32-bit while others are built 64-bit - then we need to modex
* regardless of any other consideration. The user is reqd to tell us via a
* cmd line option if this situation exists, which will result in an mca param
* being set for us, so all we need to do is check for the global boolean
* that corresponds to that param
*
* Case 2: the nodes are hetero, but the app binaries were built
* the same - i.e., either they are both 32-bit, or they are both 64-bit, but
* no mixing of the two. In this case, we include the info in the modex
*/
if (orte_hetero_apps || !orte_homogeneous_nodes) {
OPAL_OUTPUT_VERBOSE((1, orte_grpcomm_base_output,
"%s grpcomm:hier: modex is required",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME)));
if (ORTE_SUCCESS != (rc = do_modex(procs))) {
ORTE_ERROR_LOG(rc);
}
return rc;
}
}
/* see if a profile file was given to us */
if (NULL == opal_profile_file) {
/* if we don't have any other way to do this, then let's default to doing the
* modex so we at least can function, even if it isn't as fast as we might like
*/
OPAL_OUTPUT_VERBOSE((1, orte_grpcomm_base_output,
"%s grpcomm:hier: modex is required",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME)));
if (ORTE_SUCCESS != (rc = do_modex(procs))) {
ORTE_ERROR_LOG(rc);
}
return rc;
}
fd = open(opal_profile_file, O_RDONLY);
if (fd < 0) {
orte_show_help("help-orte-runtime.txt", "grpcomm-hier:file-cant-open", true, opal_profile_file);
return ORTE_ERR_NOT_FOUND;
}
OPAL_OUTPUT_VERBOSE((1, orte_grpcomm_base_output,
"%s grpcomm:hier:modex reading %s file",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME), opal_profile_file));
/* loop through file until end */
boptr = &bo;
nd = (orte_nid_t**)orte_nidmap.addr;
while (0 < read(fd, &bo.size, sizeof(bo.size))) {
/* this is the number of bytes in the byte object */
bo.bytes = malloc(bo.size);
if (0 > read(fd, bo.bytes, bo.size)) {
orte_show_help("help-orte-runtime.txt", "orte_nidmap:unable-read-file", true, opal_profile_file);
close(fd);
return ORTE_ERR_FILE_READ_FAILURE;
}
/* load the byte object into a buffer for unpacking */
OBJ_CONSTRUCT(&bobuf, opal_buffer_t);
opal_dss.load(&bobuf, boptr->bytes, boptr->size);
/* unpack the nodename */
n = 1;
if (ORTE_SUCCESS != (rc = opal_dss.unpack(&bobuf, &nodename, &n, OPAL_STRING))) {
ORTE_ERROR_LOG(rc);
return rc;
}
/* find this node in nidmap */
for (i=0, ndptr=NULL; i < orte_nidmap.size && NULL != nd[i]; i++) {
/* since we may not have kept fqdn hostnames, we can only check
* for equality to the length of the name in the nid
*/
if (0 == strncmp(nd[i]->name, nodename, strlen(nd[i]->name))) {
ndptr = nd[i];
break;
}
}
free(nodename); /* done with this */
if (NULL == ndptr) {
/* didn't find it! */
ORTE_ERROR_LOG(ORTE_ERR_NOT_FOUND);
return ORTE_ERR_NOT_FOUND;
}
/* loop through the rest of the object to unpack the attr's themselves */
n = 1;
while (ORTE_SUCCESS == opal_dss.unpack(&bobuf, &attr, &n, OPAL_STRING)) {
attrdata = OBJ_NEW(orte_attr_t);
attrdata->name = strdup(attr);
/* read the number of bytes in the blob */
n = 1;
if (ORTE_SUCCESS != (rc = opal_dss.unpack(&bobuf, &attrdata->size, &n, OPAL_INT32))) {
ORTE_ERROR_LOG(rc);
return rc;
}
/* unpack the bytes */
attrdata->bytes = malloc(attrdata->size);
if (ORTE_SUCCESS != (rc = opal_dss.unpack(&bobuf, attrdata->bytes, &attrdata->size, OPAL_BYTE))) {
ORTE_ERROR_LOG(rc);
return rc;
}
/* add to our list for this node */
opal_list_append(&ndptr->attrs, &attrdata->super);
}
OBJ_DESTRUCT(&bobuf);
}
OPAL_OUTPUT_VERBOSE((1, orte_grpcomm_base_output,
"%s grpcomm:hier: modex completed",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME)));
return rc;
}
/* the HNP will -never- execute the following as it is NOT an MPI process */
static int set_proc_attr(const char *attr_name, const void *data, size_t size)
{
OPAL_OUTPUT_VERBOSE((1, orte_grpcomm_base_output,
"%s grpcomm:hier:set_proc_attr for attribute %s",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME), attr_name));
/* we always have to set our own attributes in case they are needed for
* a connect/accept at some later time
*/
return orte_grpcomm_base_set_proc_attr(attr_name, data, size);
}
static int get_proc_attr(const orte_process_name_t proc,
const char * attribute_name, void **val,
size_t *size)
{
orte_nid_t *nid;
opal_list_item_t *item;
orte_attr_t *attr;
/* find this proc's node in the nidmap */
if (NULL == (nid = orte_util_lookup_nid((orte_process_name_t*)&proc))) {
/* proc wasn't found - return error */
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:hier:get_proc_attr: no modex entry for proc %s",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
ORTE_NAME_PRINT(&proc)));
return ORTE_ERR_NOT_FOUND;
}
/* look for this attribute */
for (item = opal_list_get_first(&nid->attrs);
item != opal_list_get_end(&nid->attrs);
item = opal_list_get_next(item)) {
attr = (orte_attr_t*)item;
if (0 == strcmp(attr->name, attribute_name)) {
/* copy the data to the caller */
void *copy = malloc(attr->size);
if (copy == NULL) {
return ORTE_ERR_OUT_OF_RESOURCE;
}
memcpy(copy, attr->bytes, attr->size);
*val = copy;
*size = attr->size;
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:hier:get_proc_attr: found %d bytes for attr %s on proc %s",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME), (int)attr->size,
attribute_name, ORTE_NAME_PRINT(&proc)));
return ORTE_SUCCESS;
}
}
/* get here if attribute isn't found */
OPAL_OUTPUT_VERBOSE((5, orte_grpcomm_base_output,
"%s grpcomm:hier:get_proc_attr: no attr avail or zero byte size for proc %s attribute %s",
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
ORTE_NAME_PRINT(&proc), attribute_name));
*val = NULL;
*size = 0;
return ORTE_SUCCESS;
}