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

893 строки
33 KiB
C
Исходник Ответственный История

/*
* Copyright (c) 2004-2006 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2011 The University of Tennessee and The University
* of Tennessee Research Foundation. All rights
* reserved.
* Copyright (c) 2004-2006 High Performance Computing Center Stuttgart,
* University of Stuttgart. All rights reserved.
* Copyright (c) 2004-2006 The Regents of the University of California.
* All rights reserved.
* Copyright (c) 2006-2007 Cisco Systems, Inc. All rights reserved.
* Copyright (c) 2012 Los Alamos National Security, LLC. All rights
* reserved.
* Copyright (c) 2013-2014 Intel, Inc. All rights reserved
* $COPYRIGHT$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#include "ompi_config.h"
#include <string.h>
#include <strings.h>
#include "ompi/constants.h"
#include "opal/datatype/opal_convertor.h"
#include "opal/threads/mutex.h"
#include "opal/dss/dss.h"
#include "opal/util/arch.h"
#include "opal/util/show_help.h"
#include "opal/mca/dstore/dstore.h"
#include "opal/mca/hwloc/base/base.h"
#include "ompi/proc/proc.h"
#include "ompi/datatype/ompi_datatype.h"
#include "ompi/runtime/mpiruntime.h"
#include "ompi/runtime/ompi_module_exchange.h"
#include "ompi/runtime/params.h"
static opal_list_t ompi_proc_list;
static opal_mutex_t ompi_proc_lock;
ompi_proc_t* ompi_proc_local_proc = NULL;
static void ompi_proc_construct(ompi_proc_t* proc);
static void ompi_proc_destruct(ompi_proc_t* proc);
OBJ_CLASS_INSTANCE(
ompi_proc_t,
opal_list_item_t,
ompi_proc_construct,
ompi_proc_destruct
);
void ompi_proc_construct(ompi_proc_t* proc)
{
bzero(proc->proc_endpoints, sizeof(proc->proc_endpoints));
/* By default all processors are supposedly having the same architecture as me. Thus,
* by default we run in a homogeneous environment. Later, when the RTE can tell us
* the arch of the remote nodes, we will have to set the convertors to the correct
* architecture.
*/
proc->proc_arch = opal_local_arch;
proc->proc_convertor = ompi_mpi_local_convertor;
OBJ_RETAIN( ompi_mpi_local_convertor );
proc->proc_flags = 0;
/* initialize this pointer to NULL */
proc->proc_hostname = NULL;
}
void ompi_proc_destruct(ompi_proc_t* proc)
{
/* As all the convertors are created with OBJ_NEW we can just call OBJ_RELEASE. All, except
* the local convertor, will get destroyed at some point here. If the reference count is correct
* the local convertor (who has the reference count increased in the datatype) will not get
* destroyed here. It will be destroyed later when the ompi_datatype_finalize is called.
*/
OBJ_RELEASE( proc->proc_convertor );
if (NULL != proc->proc_hostname) {
free(proc->proc_hostname);
}
OPAL_THREAD_LOCK(&ompi_proc_lock);
opal_list_remove_item(&ompi_proc_list, (opal_list_item_t*)proc);
OPAL_THREAD_UNLOCK(&ompi_proc_lock);
}
int ompi_proc_init(void)
{
ompi_vpid_t i;
#if OPAL_ENABLE_HETEROGENEOUS_SUPPORT
int ret;
#endif
OBJ_CONSTRUCT(&ompi_proc_list, opal_list_t);
OBJ_CONSTRUCT(&ompi_proc_lock, opal_mutex_t);
/* create proc structures and find self */
for( i = 0; i < ompi_process_info.num_procs; i++ ) {
ompi_proc_t *proc = OBJ_NEW(ompi_proc_t);
opal_list_append(&ompi_proc_list, (opal_list_item_t*)proc);
proc->proc_name.jobid = OMPI_PROC_MY_NAME->jobid;
proc->proc_name.vpid = i;
if (i == OMPI_PROC_MY_NAME->vpid) {
ompi_proc_local_proc = proc;
proc->proc_flags = OPAL_PROC_ALL_LOCAL;
proc->proc_hostname = strdup(ompi_process_info.nodename);
proc->proc_arch = opal_local_arch;
#if OPAL_ENABLE_HETEROGENEOUS_SUPPORT
/* add our arch to the modex */
if (OMPI_SUCCESS != (ret = ompi_modex_send_key_value("OMPI_ARCH", &proc->proc_arch, OPAL_UINT32))) {
return ret;
}
#endif
}
}
return OMPI_SUCCESS;
}
int ompi_proc_set_locality(ompi_proc_t *proc)
{
opal_hwloc_locality_t locality;
ompi_vpid_t vpid;
int ret;
opal_list_t myvals;
opal_value_t *kv, kvn;
/* get the locality information - do not use modex recv for
* this request as that will automatically cause the hostname
* to be loaded as well
*/
OBJ_CONSTRUCT(&myvals, opal_list_t);
if (OMPI_SUCCESS == opal_dstore.fetch(opal_dstore_internal,
(opal_identifier_t*)&proc->proc_name,
OPAL_DSTORE_LOCALITY, &myvals)) {
kv = (opal_value_t*)opal_list_get_first(&myvals);
proc->proc_flags = kv->data.uint16;
OPAL_LIST_DESTRUCT(&myvals);
return OMPI_SUCCESS;
}
OPAL_LIST_DESTRUCT(&myvals);
/* if we don't already have it, compute and save it for future use */
OBJ_CONSTRUCT(&myvals, opal_list_t);
if (OMPI_SUCCESS != (ret = opal_dstore.fetch(opal_dstore_nonpeer,
(opal_identifier_t*)&proc->proc_name,
OMPI_RTE_NODE_ID, &myvals))) {
OPAL_LIST_DESTRUCT(&myvals);
return ret;
}
kv = (opal_value_t*)opal_list_get_first(&myvals);
vpid = kv->data.uint32;
OPAL_LIST_DESTRUCT(&myvals);
/* if we are on different nodes, then we are probably non-local */
if (vpid != OMPI_RTE_MY_NODEID) {
locality = OPAL_PROC_NON_LOCAL;
#ifdef OMPI_RTE_HOST_ID
/* see if coprocessors were detected - if the hostid isn't
* present, then no coprocessors were detected and we can
* ignore this test
*/
OBJ_CONSTRUCT(&myvals, opal_list_t);
if (OMPI_SUCCESS == opal_dstore.fetch(opal_dstore_internal,
(opal_identifier_t*)&proc->proc_name,
OMPI_RTE_HOST_ID, &myvals)) {
kv = (opal_value_t*)opal_list_get_first(&myvals);
vpid = kv->data.uint32;
/* if this matches my host id, then we are on the same host,
* but not on the same board
*/
if (vpid == ompi_process_info.my_hostid) {
locality = OPAL_PROC_ON_HOST;
} else {
locality = OPAL_PROC_NON_LOCAL;
}
}
OPAL_LIST_DESTRUCT(&myvals);
#endif
} else {
#if OPAL_HAVE_HWLOC
{
char *cpu_bitmap;
/* retrieve the binding for the other proc */
OBJ_CONSTRUCT(&myvals, opal_list_t);
if (OMPI_SUCCESS != (ret = opal_dstore.fetch(opal_dstore_internal,
(opal_identifier_t*)&proc->proc_name,
OPAL_DSTORE_CPUSET, &myvals))) {
/* check the nonpeer data in case of comm_spawn */
ret = opal_dstore.fetch(opal_dstore_nonpeer,
(opal_identifier_t*)&proc->proc_name,
OPAL_DSTORE_CPUSET, &myvals);
}
if (OMPI_SUCCESS != ret) {
/* we don't know their cpuset, so nothing more we can say */
locality = OPAL_PROC_ON_NODE;
} else {
kv = (opal_value_t*)opal_list_get_first(&myvals);
cpu_bitmap = kv->data.string;
if (NULL == cpu_bitmap || NULL == ompi_process_info.cpuset) {
/* one or both of us is not bound, so all we can say is we are on the
* same node
*/
locality = OPAL_PROC_ON_NODE;
} else {
/* we share a node - see what else we share */
locality = opal_hwloc_base_get_relative_locality(opal_hwloc_topology,
ompi_process_info.cpuset,
cpu_bitmap);
}
}
OPAL_LIST_DESTRUCT(&myvals);
}
#else
/* all we know is that we share this node */
locality = OPAL_PROC_ON_NODE;
#endif
}
OBJ_CONSTRUCT(&kvn, opal_value_t);
kvn.key = strdup(OPAL_DSTORE_LOCALITY);
kvn.type = OPAL_HWLOC_LOCALITY_T;
kvn.data.uint16 = locality;
ret = opal_dstore.store(opal_dstore_internal, (opal_identifier_t*)&proc, &kvn);
OBJ_DESTRUCT(&kvn);
/* set the proc's local value as well */
proc->proc_flags = locality;
return ret;
}
/**
* The process creation is split into two steps. The second step
* is the important one, it sets the properties of the remote
* process, such as architecture, node name and locality flags.
*
* This function is to be called __only__ after the modex exchange
* has been performed, in order to allow the modex to carry the data
* instead of requiring the runtime to provide it.
*/
int ompi_proc_complete_init(void)
{
ompi_proc_t *proc = NULL;
opal_list_item_t *item = NULL;
int ret, errcode = OMPI_SUCCESS;
OPAL_THREAD_LOCK(&ompi_proc_lock);
for( item = opal_list_get_first(&ompi_proc_list);
item != opal_list_get_end(&ompi_proc_list);
item = opal_list_get_next(item)) {
proc = (ompi_proc_t*)item;
if (proc->proc_name.vpid != OMPI_PROC_MY_NAME->vpid) {
/* get the locality information */
ret = ompi_proc_set_locality(proc);
if (OMPI_SUCCESS != ret) {
errcode = ret;
break;
}
if (ompi_process_info.num_procs < ompi_hostname_cutoff) {
/* IF the number of procs falls below the specified cutoff,
* then we assume the job is small enough that retrieving
* the hostname (which will typically cause retrieval of
* ALL modex info for this proc) will have no appreciable
* impact on launch scaling
*/
ret = ompi_modex_recv_key_value(OMPI_DB_HOSTNAME, proc, (void**)&(proc->proc_hostname), OPAL_STRING);
if (OMPI_SUCCESS != ret) {
errcode = ret;
break;
}
} else {
/* just set the hostname to NULL for now - we'll fill it in
* as modex_recv's are called for procs we will talk to, thus
* avoiding retrieval of ALL modex info for this proc until
* required. Transports that delay calling modex_recv until
* first message will therefore scale better than those that
* call modex_recv on all procs during init.
*/
proc->proc_hostname = NULL;
}
#if OPAL_ENABLE_HETEROGENEOUS_SUPPORT
/* get the remote architecture */
{
uint32_t *ui32ptr;
ui32ptr = &(proc->proc_arch);
ret = ompi_modex_recv_key_value("OMPI_ARCH", proc, (void**)&ui32ptr, OPAL_UINT32);
if (OMPI_SUCCESS == ret) {
/* if arch is different than mine, create a new convertor for this proc */
if (proc->proc_arch != opal_local_arch) {
OBJ_RELEASE(proc->proc_convertor);
proc->proc_convertor = opal_convertor_create(proc->proc_arch, 0);
}
} else if (OMPI_ERR_NOT_IMPLEMENTED == ret) {
proc->proc_arch = opal_local_arch;
} else {
errcode = ret;
break;
}
}
#else
/* must be same arch as my own */
proc->proc_arch = opal_local_arch;
#endif
}
}
OPAL_THREAD_UNLOCK(&ompi_proc_lock);
return errcode;
}
int ompi_proc_finalize (void)
{
opal_list_item_t *item;
/* remove all items from list and destroy them. Since we cannot know
* the reference count of the procs for certain, it is possible that
* a single OBJ_RELEASE won't drive the count to zero, and hence will
* not release the memory. Accordingly, we cycle through the list here,
* calling release on each item.
*
* This will cycle until it forces the reference count of each item
* to zero, thus causing the destructor to run - which will remove
* the item from the list!
*
* We cannot do this under the thread lock as the destructor will
* call it when removing the item from the list. However, this function
* is ONLY called from MPI_Finalize, and all threads are prohibited from
* calling an MPI function once ANY thread has called MPI_Finalize. Of
* course, multiple threads are allowed to call MPI_Finalize, so this
* function may get called multiple times by various threads. We believe
* it is thread safe to do so...though it may not -appear- to be so
* without walking through the entire list/destructor sequence.
*/
while (opal_list_get_end(&ompi_proc_list) != (item = opal_list_get_first(&ompi_proc_list))) {
OBJ_RELEASE(item);
}
/* now destruct the list and thread lock */
OBJ_DESTRUCT(&ompi_proc_list);
OBJ_DESTRUCT(&ompi_proc_lock);
return OMPI_SUCCESS;
}
ompi_proc_t** ompi_proc_world(size_t *size)
{
ompi_proc_t **procs;
ompi_proc_t *proc;
size_t count = 0;
ompi_rte_cmp_bitmask_t mask;
ompi_process_name_t my_name;
/* check bozo case */
if (NULL == ompi_proc_local_proc) {
return NULL;
}
mask = OMPI_RTE_CMP_JOBID;
my_name = ompi_proc_local_proc->proc_name;
/* First count how many match this jobid */
OPAL_THREAD_LOCK(&ompi_proc_lock);
for (proc = (ompi_proc_t*)opal_list_get_first(&ompi_proc_list);
proc != (ompi_proc_t*)opal_list_get_end(&ompi_proc_list);
proc = (ompi_proc_t*)opal_list_get_next(proc)) {
if (OPAL_EQUAL == ompi_rte_compare_name_fields(mask, &proc->proc_name, &my_name)) {
++count;
}
}
/* allocate an array */
procs = (ompi_proc_t**) malloc(count * sizeof(ompi_proc_t*));
if (NULL == procs) {
OPAL_THREAD_UNLOCK(&ompi_proc_lock);
return NULL;
}
/* now save only the procs that match this jobid */
count = 0;
for (proc = (ompi_proc_t*)opal_list_get_first(&ompi_proc_list);
proc != (ompi_proc_t*)opal_list_get_end(&ompi_proc_list);
proc = (ompi_proc_t*)opal_list_get_next(proc)) {
if (OPAL_EQUAL == ompi_rte_compare_name_fields(mask, &proc->proc_name, &my_name)) {
/* DO NOT RETAIN THIS OBJECT - the reference count on this
* object will be adjusted by external callers. The intent
* here is to allow the reference count to drop to zero if
* the app no longer desires to communicate with this proc.
* For example, the proc may call comm_disconnect on all
* communicators involving this proc. In such cases, we want
* the proc object to be removed from the list. By not incrementing
* the reference count here, we allow this to occur.
*
* We don't implement that yet, but we are still safe for now as
* the OBJ_NEW in ompi_proc_init owns the initial reference
* count which cannot be released until ompi_proc_finalize is
* called.
*/
procs[count++] = proc;
}
}
OPAL_THREAD_UNLOCK(&ompi_proc_lock);
*size = count;
return procs;
}
ompi_proc_t** ompi_proc_all(size_t* size)
{
ompi_proc_t **procs =
(ompi_proc_t**) malloc(opal_list_get_size(&ompi_proc_list) * sizeof(ompi_proc_t*));
ompi_proc_t *proc;
size_t count = 0;
if (NULL == procs) {
return NULL;
}
OPAL_THREAD_LOCK(&ompi_proc_lock);
for(proc = (ompi_proc_t*)opal_list_get_first(&ompi_proc_list);
proc != (ompi_proc_t*)opal_list_get_end(&ompi_proc_list);
proc = (ompi_proc_t*)opal_list_get_next(proc)) {
/* We know this isn't consistent with the behavior in ompi_proc_world,
* but we are leaving the RETAIN for now because the code using this function
* assumes that the results need to be released when done. It will
* be cleaned up later as the "fix" will impact other places in
* the code
*/
OBJ_RETAIN(proc);
procs[count++] = proc;
}
OPAL_THREAD_UNLOCK(&ompi_proc_lock);
*size = count;
return procs;
}
ompi_proc_t** ompi_proc_self(size_t* size)
{
ompi_proc_t **procs = (ompi_proc_t**) malloc(sizeof(ompi_proc_t*));
if (NULL == procs) {
return NULL;
}
/* We know this isn't consistent with the behavior in ompi_proc_world,
* but we are leaving the RETAIN for now because the code using this function
* assumes that the results need to be released when done. It will
* be cleaned up later as the "fix" will impact other places in
* the code
*/
OBJ_RETAIN(ompi_proc_local_proc);
*procs = ompi_proc_local_proc;
*size = 1;
return procs;
}
ompi_proc_t * ompi_proc_find ( const ompi_process_name_t * name )
{
ompi_proc_t *proc, *rproc=NULL;
ompi_rte_cmp_bitmask_t mask;
/* return the proc-struct which matches this jobid+process id */
mask = OMPI_RTE_CMP_JOBID | OMPI_RTE_CMP_VPID;
OPAL_THREAD_LOCK(&ompi_proc_lock);
for(proc = (ompi_proc_t*)opal_list_get_first(&ompi_proc_list);
proc != (ompi_proc_t*)opal_list_get_end(&ompi_proc_list);
proc = (ompi_proc_t*)opal_list_get_next(proc)) {
if (OPAL_EQUAL == ompi_rte_compare_name_fields(mask, &proc->proc_name, name)) {
rproc = proc;
break;
}
}
OPAL_THREAD_UNLOCK(&ompi_proc_lock);
return rproc;
}
int ompi_proc_refresh(void) {
ompi_proc_t *proc = NULL;
opal_list_item_t *item = NULL;
ompi_vpid_t i = 0;
int ret=OMPI_SUCCESS;
OPAL_THREAD_LOCK(&ompi_proc_lock);
for( item = opal_list_get_first(&ompi_proc_list), i = 0;
item != opal_list_get_end(&ompi_proc_list);
item = opal_list_get_next(item), ++i ) {
proc = (ompi_proc_t*)item;
/* Does not change: proc->proc_name.vpid */
proc->proc_name.jobid = OMPI_PROC_MY_NAME->jobid;
/* Make sure to clear the local flag before we set it below */
proc->proc_flags = 0;
if (i == OMPI_PROC_MY_NAME->vpid) {
ompi_proc_local_proc = proc;
proc->proc_flags = OPAL_PROC_ALL_LOCAL;
proc->proc_hostname = ompi_process_info.nodename;
proc->proc_arch = opal_local_arch;
} else {
/* get the locality information */
ret = ompi_proc_set_locality(proc);
if (OMPI_SUCCESS != ret) {
break;
}
if (ompi_process_info.num_procs < ompi_hostname_cutoff) {
/* IF the number of procs falls below the specified cutoff,
* then we assume the job is small enough that retrieving
* the hostname (which will typically cause retrieval of
* ALL modex info for this proc) will have no appreciable
* impact on launch scaling
*/
ret = ompi_modex_recv_key_value(OMPI_DB_HOSTNAME, proc, (void**)&(proc->proc_hostname), OPAL_STRING);
if (OMPI_SUCCESS != ret) {
break;
}
} else {
/* just set the hostname to NULL for now - we'll fill it in
* as modex_recv's are called for procs we will talk to, thus
* avoiding retrieval of ALL modex info for this proc until
* required. Transports that delay calling modex_recv until
* first message will therefore scale better than those that
* call modex_recv on all procs during init.
*/
proc->proc_hostname = NULL;
}
#if OPAL_ENABLE_HETEROGENEOUS_SUPPORT
{
/* get the remote architecture */
uint32_t* uiptr = &(proc->proc_arch);
ret = ompi_modex_recv_key_value("OMPI_ARCH", proc, (void**)&uiptr, OPAL_UINT32);
if (OMPI_SUCCESS != ret) {
break;
}
/* if arch is different than mine, create a new convertor for this proc */
if (proc->proc_arch != opal_local_arch) {
OBJ_RELEASE(proc->proc_convertor);
proc->proc_convertor = opal_convertor_create(proc->proc_arch, 0);
}
}
#else
/* must be same arch as my own */
proc->proc_arch = opal_local_arch;
#endif
}
}
OPAL_THREAD_UNLOCK(&ompi_proc_lock);
return ret;
}
int
ompi_proc_pack(ompi_proc_t **proclist, int proclistsize,
bool full_info,
opal_buffer_t* buf)
{
int i, rc;
OPAL_THREAD_LOCK(&ompi_proc_lock);
/* cycle through the provided array, packing the OMPI level
* data for each proc. This data may or may not be included
* in any subsequent modex operation, so we include it here
* to ensure completion of a connect/accept handshake. See
* the ompi/mca/dpm framework for an example of where and how
* this info is used.
*
* Eventually, we will review the procedures that call this
* function to see if duplication of communication can be
* reduced. For now, just go ahead and pack the info so it
* can be sent.
*/
for (i=0; i<proclistsize; i++) {
rc = opal_dss.pack(buf, &(proclist[i]->proc_name), 1, OMPI_NAME);
if(rc != OPAL_SUCCESS) {
OMPI_ERROR_LOG(rc);
OPAL_THREAD_UNLOCK(&ompi_proc_lock);
return rc;
}
if (full_info) {
int32_t num_entries;
opal_value_t *kv;
opal_list_t data;
/* fetch all global info we know about the peer - while
* the remote procs may already know some of it, we cannot
* be certain they do. So we must include a full dump of
* everything we know about this proc, excluding INTERNAL
* data that each process computes about its peers
*/
OBJ_CONSTRUCT(&data, opal_list_t);
rc = opal_dstore.fetch(opal_dstore_peer,
(opal_identifier_t*)&proclist[i]->proc_name,
NULL, &data);
if (OPAL_SUCCESS != rc) {
OMPI_ERROR_LOG(rc);
num_entries = 0;
} else {
rc = opal_dstore.fetch(opal_dstore_nonpeer,
(opal_identifier_t*)&proclist[i]->proc_name,
NULL, &data);
if (OPAL_SUCCESS != rc) {
OMPI_ERROR_LOG(rc);
num_entries = 0;
} else {
/* count the number of entries we will send */
num_entries = opal_list_get_size(&data);
}
}
/* put the number of entries into the buffer */
rc = opal_dss.pack(buf, &num_entries, 1, OPAL_INT32);
if (OPAL_SUCCESS != rc) {
OMPI_ERROR_LOG(rc);
break;
}
/* if there are entries, store them */
while (NULL != (kv = (opal_value_t*)opal_list_remove_first(&data))) {
if (OPAL_SUCCESS != (rc = opal_dss.pack(buf, &kv, 1, OPAL_VALUE))) {
OMPI_ERROR_LOG(rc);
break;
}
OBJ_RELEASE(kv);
}
OBJ_DESTRUCT(&data);
} else {
rc = opal_dss.pack(buf, &(proclist[i]->proc_arch), 1, OPAL_UINT32);
if(rc != OPAL_SUCCESS) {
OMPI_ERROR_LOG(rc);
OPAL_THREAD_UNLOCK(&ompi_proc_lock);
return rc;
}
rc = opal_dss.pack(buf, &(proclist[i]->proc_hostname), 1, OPAL_STRING);
if(rc != OPAL_SUCCESS) {
OMPI_ERROR_LOG(rc);
OPAL_THREAD_UNLOCK(&ompi_proc_lock);
return rc;
}
}
}
OPAL_THREAD_UNLOCK(&ompi_proc_lock);
return OMPI_SUCCESS;
}
static ompi_proc_t *
ompi_proc_find_and_add(const ompi_process_name_t * name, bool* isnew)
{
ompi_proc_t *proc, *rproc = NULL;
ompi_rte_cmp_bitmask_t mask;
/* return the proc-struct which matches this jobid+process id */
mask = OMPI_RTE_CMP_JOBID | OMPI_RTE_CMP_VPID;
OPAL_THREAD_LOCK(&ompi_proc_lock);
for(proc = (ompi_proc_t*)opal_list_get_first(&ompi_proc_list);
proc != (ompi_proc_t*)opal_list_get_end(&ompi_proc_list);
proc = (ompi_proc_t*)opal_list_get_next(proc)) {
if (OPAL_EQUAL == ompi_rte_compare_name_fields(mask, &proc->proc_name, name)) {
rproc = proc;
*isnew = false;
break;
}
}
/* if we didn't find this proc in the list, create a new
* proc_t and append it to the list
*/
if (NULL == rproc) {
*isnew = true;
rproc = OBJ_NEW(ompi_proc_t);
if (NULL != rproc) {
opal_list_append(&ompi_proc_list, (opal_list_item_t*)rproc);
rproc->proc_name = *name;
}
/* caller had better fill in the rest of the proc, or there's
going to be pain later... */
}
OPAL_THREAD_UNLOCK(&ompi_proc_lock);
return rproc;
}
int
ompi_proc_unpack(opal_buffer_t* buf,
int proclistsize, ompi_proc_t ***proclist,
bool full_info,
int *newproclistsize, ompi_proc_t ***newproclist)
{
int i;
size_t newprocs_len = 0;
ompi_proc_t **plist=NULL, **newprocs = NULL;
opal_list_t myvals;
opal_value_t *kv;
/* do not free plist *ever*, since it is used in the remote group
structure of a communicator */
plist = (ompi_proc_t **) calloc (proclistsize, sizeof (ompi_proc_t *));
if ( NULL == plist ) {
return OMPI_ERR_OUT_OF_RESOURCE;
}
/* free this on the way out */
newprocs = (ompi_proc_t **) calloc (proclistsize, sizeof (ompi_proc_t *));
if (NULL == newprocs) {
free(plist);
return OMPI_ERR_OUT_OF_RESOURCE;
}
/* cycle through the array of provided procs and unpack
* their info - as packed by ompi_proc_pack
*/
for ( i=0; i<proclistsize; i++ ){
int32_t count=1;
ompi_process_name_t new_name;
uint32_t new_arch;
char *new_hostname;
bool isnew = false;
int rc;
rc = opal_dss.unpack(buf, &new_name, &count, OMPI_NAME);
if (rc != OPAL_SUCCESS) {
OMPI_ERROR_LOG(rc);
free(plist);
free(newprocs);
return rc;
}
if (!full_info) {
rc = opal_dss.unpack(buf, &new_arch, &count, OPAL_UINT32);
if (rc != OPAL_SUCCESS) {
OMPI_ERROR_LOG(rc);
free(plist);
free(newprocs);
return rc;
}
rc = opal_dss.unpack(buf, &new_hostname, &count, OPAL_STRING);
if (rc != OPAL_SUCCESS) {
OMPI_ERROR_LOG(rc);
free(plist);
free(newprocs);
return rc;
}
}
/* see if this proc is already on our ompi_proc_list */
plist[i] = ompi_proc_find_and_add(&new_name, &isnew);
if (isnew) {
/* if not, then it was added, so update the values
* in the proc_t struct with the info that was passed
* to us
*/
newprocs[newprocs_len++] = plist[i];
if (full_info) {
int32_t num_recvd_entries;
int32_t cnt;
int32_t j;
/* unpack the number of entries for this proc */
cnt = 1;
if (OPAL_SUCCESS != (rc = opal_dss.unpack(buf, &num_recvd_entries, &cnt, OPAL_INT32))) {
OMPI_ERROR_LOG(rc);
break;
}
/*
* Extract the attribute names and values
*/
for (j = 0; j < num_recvd_entries; j++) {
cnt = 1;
if (OPAL_SUCCESS != (rc = opal_dss.unpack(buf, &kv, &cnt, OPAL_VALUE))) {
OMPI_ERROR_LOG(rc);
break;
}
/* if this is me, ignore the data - we already have it in the db */
if (OPAL_EQUAL != ompi_rte_compare_name_fields(OMPI_RTE_CMP_ALL,
OMPI_PROC_MY_NAME, &new_name)) {
/* store it in the database */
if (OPAL_SUCCESS != (rc = opal_dstore.store(opal_dstore_peer,
(opal_identifier_t*)&new_name, kv))) {
OMPI_ERROR_LOG(rc);
}
}
OBJ_RELEASE(kv);
}
#if OPAL_ENABLE_HETEROGENEOUS_SUPPORT
OBJ_CONSTRUCT(&myvals, opal_list_t);
rc = opal_dstore.fetch(opal_dstore_peer,
(opal_identifier_t*)&new_name,
"OMPI_ARCH", &myvals);
if( OPAL_SUCCESS == rc ) {
kv = (opal_value_t*)opal_list_get_first(&myvals);
new_arch = kv->data.uint32;
} else {
new_arch = opal_local_arch;
}
OPAL_LIST_DESTRUCT(&myvals);
#else
new_arch = opal_local_arch;
#endif
if (ompi_process_info.num_procs < ompi_hostname_cutoff) {
/* retrieve the hostname */
OBJ_CONSTRUCT(&myvals, opal_list_t);
rc = opal_dstore.fetch(opal_dstore_peer,
(opal_identifier_t*)&new_name,
OMPI_DB_HOSTNAME, &myvals);
if( OPAL_SUCCESS == rc ) {
kv = (opal_value_t*)opal_list_get_first(&myvals);
new_hostname = strdup(kv->data.string);
} else {
new_hostname = NULL;
}
OPAL_LIST_DESTRUCT(&myvals);
} else {
/* just set the hostname to NULL for now - we'll fill it in
* as modex_recv's are called for procs we will talk to
*/
new_hostname = NULL;
}
}
/* update all the values */
plist[i]->proc_arch = new_arch;
/* if arch is different than mine, create a new convertor for this proc */
if (plist[i]->proc_arch != opal_local_arch) {
#if OPAL_ENABLE_HETEROGENEOUS_SUPPORT
OBJ_RELEASE(plist[i]->proc_convertor);
plist[i]->proc_convertor = opal_convertor_create(plist[i]->proc_arch, 0);
#else
opal_show_help("help-mpi-runtime",
"heterogeneous-support-unavailable",
true, ompi_process_info.nodename,
new_hostname == NULL ? "<hostname unavailable>" :
new_hostname);
free(plist);
free(newprocs);
return OMPI_ERR_NOT_SUPPORTED;
#endif
}
if (0 == strcmp(ompi_proc_local_proc->proc_hostname, new_hostname)) {
plist[i]->proc_flags |= (OPAL_PROC_ON_NODE | OPAL_PROC_ON_CU | OPAL_PROC_ON_CLUSTER);
}
/* Save the hostname */
plist[i]->proc_hostname = new_hostname;
} else {
if (full_info) {
int32_t num_recvd_entries;
int32_t j, cnt;
/* discard all keys: they are already locally known */
cnt = 1;
if (OPAL_SUCCESS == (rc = opal_dss.unpack(buf, &num_recvd_entries, &cnt, OPAL_INT32))) {
for (j = 0; j < num_recvd_entries; j++) {
opal_value_t *kv;
cnt = 1;
if (OPAL_SUCCESS != (rc = opal_dss.unpack(buf, &kv, &cnt, OPAL_VALUE))) {
OMPI_ERROR_LOG(rc);
continue;
}
OBJ_RELEASE(kv);
}
} else {
OMPI_ERROR_LOG(rc);
}
}
}
}
if (NULL != newproclistsize) *newproclistsize = newprocs_len;
if (NULL != newproclist) {
*newproclist = newprocs;
} else if (newprocs != NULL) {
free(newprocs);
}
*proclist = plist;
return OMPI_SUCCESS;
}
Показать git blame Копировать постоянную ссылку