ports/openmpi
ports/openmpi
1
1
Форкнуть 0
Код Релизы Активность
openmpi/ompi/mca/btl/udapl/btl_udapl_component.c
Ralph Castain fceabb2498 Update libevent to the 2.0 series, currently at 2.0.7rc. We will update to their final release when it becomes available. Currently known errors exist in unused portions of the libevent code. This revision passes the IBM test suite on a Linux machine and on a standalone Mac. 
This is a fairly intrusive change, but outside of the moving of opal/event to opal/mca/event, the only changes involved (a) changing all calls to opal_event functions to reflect the new framework instead, and (b) ensuring that all opal_event_t objects are properly constructed since they are now true opal_objects.

Note: Shiqing has just returned from vacation and has not yet had a chance to complete the Windows integration. Thus, this commit almost certainly breaks Windows support on the trunk. However, I want this to have a chance to soak for as long as possible before I become less available a week from today (going to be at a class for 5 days, and thus will only be sparingly available) so we can find and fix any problems.

Biggest change is moving the libevent code from opal/event to a new opal/mca/event framework. This was done to make it much easier to update libevent in the future. New versions can be inserted as a new component and tested in parallel with the current version until validated, then we can remove the earlier version if we so choose. This is a statically built framework ala installdirs, so only one component will build at a time. There is no selection logic - the sole compiled component simply loads its function pointers into the opal_event struct.

I have gone thru the code base and converted all the libevent calls I could find. However, I cannot compile nor test every environment. It is therefore quite likely that errors remain in the system. Please keep an eye open for two things:

1. compile-time errors: these will be obvious as calls to the old functions (e.g., opal_evtimer_new) must be replaced by the new framework APIs (e.g., opal_event.evtimer_new)

2. run-time errors: these will likely show up as segfaults due to missing constructors on opal_event_t objects. It appears that it became a typical practice for people to "init" an opal_event_t by simply using memset to zero it out. This will no longer work - you must either OBJ_NEW or OBJ_CONSTRUCT an opal_event_t. I tried to catch these cases, but may have missed some. Believe me, you'll know when you hit it.

There is also the issue of the new libevent "no recursion" behavior. As I described on a recent email, we will have to discuss this and figure out what, if anything, we need to do.

This commit was SVN r23925.
2010-10-24 18:35:54 +00:00

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

/* -*- Mode: C; c-basic-offset:4 ; -*- */
/*
* Copyright (c) 2004-2007 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2008 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 Sandia National Laboratories. All rights
* reserved.
* Copyright (c) 2007-2009 Sun Microsystems, Inc. All rights reserved.
* Copyright (c) 2007-2008 Cisco Systems, Inc. All rights reserved.
*
* $COPYRIGHT$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#include "ompi_config.h"
#include "ompi/constants.h"
#include "opal/mca/event/event.h"
#include "opal/util/argv.h"
#include "ompi/mca/btl/btl.h"
#include "opal/mca/base/mca_base_param.h"
#include "ompi/mca/mpool/base/base.h"
#include "ompi/mca/mpool/rdma/mpool_rdma.h"
#include "btl_udapl.h"
#include "btl_udapl_frag.h"
#include "btl_udapl_endpoint.h"
#include "btl_udapl_mca.h"
#include "btl_udapl_proc.h"
#include "ompi/mca/btl/base/base.h"
#include "ompi/mca/btl/base/btl_base_error.h"
#include "btl_udapl_endpoint.h"
#include "orte/util/proc_info.h"
#include "ompi/runtime/ompi_module_exchange.h"
#include "ompi/runtime/mpiruntime.h"
/*
* Local Functions
*/
static inline int mca_btl_udapl_frag_progress_one(mca_btl_udapl_module_t* udapl_btl,
mca_btl_udapl_frag_t* frag);
void mca_btl_udapl_frag_progress_pending(mca_btl_udapl_module_t* udapl_btl,
mca_btl_base_endpoint_t* endpoint,
const int connection);
static int mca_btl_udapl_modify_ia_list(DAT_COUNT *num_info_entries,
DAT_PROVIDER_INFO* datinfo);
static const char*
mca_btl_udapl_dat_event_to_string(DAT_EVENT_NUMBER event_number);
mca_btl_udapl_component_t mca_btl_udapl_component = {
{
/* First, the mca_base_component_t struct containing meta information
about the component itself */
{
MCA_BTL_BASE_VERSION_2_0_0,
"udapl", /* MCA component name */
OMPI_MAJOR_VERSION, /* MCA component major version */
OMPI_MINOR_VERSION, /* MCA component minor version */
OMPI_RELEASE_VERSION, /* MCA component release version */
mca_btl_udapl_component_open, /* component open */
mca_btl_udapl_component_close /* component close */
},
{
/* The component is not checkpoint ready */
MCA_BASE_METADATA_PARAM_NONE
},
mca_btl_udapl_component_init,
mca_btl_udapl_component_progress,
}
};
/*
* Predefined and fixed size structure containing DAT_EVENT values
* and associated string as defined in: "uDAPL:User Direct Access
* Programming Library v1.2 Sept 15, 2004", DAT Collaborative Organization.
*/
static struct mca_btl_udapl_dat_events {
DAT_EVENT_NUMBER value;
const char* name;
} mca_btl_udapl_dat_events[] = {
{ DAT_DTO_COMPLETION_EVENT,
"DAT_DTO_COMPLETION_EVENT" },
{ DAT_RMR_BIND_COMPLETION_EVENT,
"DAT_RMR_BIND_COMPLETION_EVENT" },
{ DAT_CONNECTION_REQUEST_EVENT,
"DAT_CONNECTION_REQUEST_EVENT" },
{ DAT_CONNECTION_EVENT_ESTABLISHED,
"DAT_CONNECTION_EVENT_ESTABLISHED" },
{ DAT_CONNECTION_EVENT_PEER_REJECTED,
"DAT_CONNECTION_EVENT_PEER_REJECTED" },
{ DAT_CONNECTION_EVENT_NON_PEER_REJECTED,
"DAT_CONNECTION_EVENT_NON_PEER_REJECTED" },
{ DAT_CONNECTION_EVENT_ACCEPT_COMPLETION_ERROR,
"DAT_CONNECTION_EVENT_ACCEPT_COMPLETION_ERROR" },
{ DAT_CONNECTION_EVENT_DISCONNECTED,
"DAT_CONNECTION_EVENT_DISCONNECTED" },
{ DAT_CONNECTION_EVENT_BROKEN,
"DAT_CONNECTION_EVENT_BROKEN" },
{ DAT_CONNECTION_EVENT_TIMED_OUT,
"DAT_CONNECTION_EVENT_TIMED_OUT" },
{ DAT_CONNECTION_EVENT_UNREACHABLE,
"DAT_CONNECTION_EVENT_UNREACHABLE" },
{ DAT_ASYNC_ERROR_EVD_OVERFLOW,
"DAT_ASYNC_ERROR_EVD_OVERFLOW" },
{ DAT_ASYNC_ERROR_IA_CATASTROPHIC,
"DAT_ASYNC_ERROR_IA_CATASTROPHIC" },
{ DAT_ASYNC_ERROR_EP_BROKEN,
"DAT_ASYNC_ERROR_EP_BROKEN" },
{ DAT_ASYNC_ERROR_TIMED_OUT,
"DAT_ASYNC_ERROR_TIMED_OUT" },
{ DAT_ASYNC_ERROR_PROVIDER_INTERNAL_ERROR,
"DAT_ASYNC_ERROR_PROVIDER_INTERNAL_ERROR" },
{ DAT_SOFTWARE_EVENT,
"DAT_SOFTWARE_EVENT" }
};
/*
* Function to convert DAT_EVENT_NUMBER into a readable string.
*
* @param event_number (IN) DAT_EVENT_NUMBER value
*
* @return event string or a string indicating
* event number is invalid
*/
static const char *
mca_btl_udapl_dat_event_to_string(DAT_EVENT_NUMBER event_number)
{
int i;
int num_events = (sizeof(mca_btl_udapl_dat_events) /
sizeof(mca_btl_udapl_dat_events[0]));
for (i = 0; i < num_events; i++) {
if (mca_btl_udapl_dat_events[i].value == event_number) {
return (mca_btl_udapl_dat_events[i].name);
}
}
return ("Unknown DAT Event Number");
}
/**
* Report a uDAPL error - for debugging
*/
#if OPAL_ENABLE_DEBUG
void
mca_btl_udapl_error(DAT_RETURN ret, char* str)
{
char* major;
char* minor;
if(DAT_SUCCESS != dat_strerror(ret,
(const char**)&major, (const char**)&minor))
{
printf("dat_strerror failed! ret is %d\n", ret);
exit(-1);
}
OPAL_OUTPUT((0, "ERROR: %s %s %s\n", str, major, minor));
}
#endif
/*
* Called by MCA framework to open the component, registers
* component parameters.
*/
int mca_btl_udapl_component_open(void)
{
int rc = OMPI_SUCCESS;
/* initialize state */
mca_btl_udapl_component.udapl_num_btls=0;
mca_btl_udapl_component.udapl_btls=NULL;
mca_btl_udapl_component.ro_aware_system=0;
/* initialize objects */
OBJ_CONSTRUCT(&mca_btl_udapl_component.udapl_procs, opal_list_t);
OBJ_CONSTRUCT(&mca_btl_udapl_component.udapl_lock, opal_mutex_t);
/* register uDAPL MCA parameters */
rc = mca_btl_udapl_register_mca_params();
/* compute udapl_eager_frag_size and udapl_max_frag_size */
mca_btl_udapl_component.udapl_eager_frag_size =
mca_btl_udapl_module.super.btl_eager_limit;
mca_btl_udapl_module.super.btl_eager_limit -=
(sizeof(mca_btl_udapl_footer_t) + sizeof(mca_btl_udapl_rdma_footer_t));
mca_btl_udapl_component.udapl_max_frag_size =
mca_btl_udapl_module.super.btl_max_send_size;
mca_btl_udapl_module.super.btl_max_send_size -=
(sizeof(mca_btl_udapl_footer_t) + sizeof(mca_btl_udapl_rdma_footer_t));
/* compute udapl_eager_rdma_frag_size */
mca_btl_udapl_component.udapl_eager_rdma_frag_size =
sizeof(mca_btl_udapl_frag_eager_rdma_t) +
mca_btl_udapl_component.udapl_eager_frag_size;
return rc;
}
/*
* component cleanup - sanity checking of queue lengths
*/
int mca_btl_udapl_component_close(void)
{
/* TODO - what needs to be done here? */
return OMPI_SUCCESS;
}
/*
* Register uDAPL component addressing information. The MCA framework
* will make this available to all peers.
*/
static int
mca_btl_udapl_modex_send(void)
{
int rc;
size_t i;
size_t size;
mca_btl_udapl_addr_t *addrs = NULL;
size = sizeof(mca_btl_udapl_addr_t) *
mca_btl_udapl_component.udapl_num_btls;
if (0 != size) {
addrs = (mca_btl_udapl_addr_t*)malloc(size);
if (NULL == addrs) {
return OMPI_ERR_OUT_OF_RESOURCE;
}
memset(addrs, 0, size);
for (i = 0; i < mca_btl_udapl_component.udapl_num_btls; i++) {
mca_btl_udapl_module_t* btl = mca_btl_udapl_component.udapl_btls[i];
addrs[i] = btl->udapl_addr;
}
}
rc = ompi_modex_send(
&mca_btl_udapl_component.super.btl_version, addrs, size);
if (NULL != addrs) {
free (addrs);
}
return rc;
}
/*
* Callback function used for udapl btl internal control messages.
*
* @param btl (IN) BTL module
* @param tag (IN) Not used but part of callback interface
* @param descriptor (IN) Description of the data that was just transferred
* @param cbdata (IN) Data used by call back function. Not used.
*
*/
static void mca_btl_udapl_receive_control(struct mca_btl_base_module_t* btl,
mca_btl_base_tag_t tag,
mca_btl_base_descriptor_t* descriptor,
void* cbdata)
{
mca_btl_udapl_frag_t* frag = (mca_btl_udapl_frag_t*)descriptor;
mca_btl_udapl_endpoint_t* endpoint = frag->endpoint;
mca_btl_udapl_control_header_t* ctl_hdr =
frag->segment.seg_addr.pval;
switch (ctl_hdr->type) {
case MCA_BTL_UDAPL_CONTROL_RDMA_CONNECT:
{
mca_btl_udapl_eager_rdma_connect_t* rdma_connect =
frag->segment.seg_addr.pval;
if (endpoint->endpoint_eager_rdma_remote.base.pval) {
BTL_ERROR(("ERROR: Received RDMA connect twice!"));
return;
}
endpoint->endpoint_eager_rdma_remote.rkey = rdma_connect->rkey;
endpoint->endpoint_eager_rdma_remote.base.pval =
rdma_connect->rdma_start.pval;
OPAL_THREAD_ADD32(&(endpoint->endpoint_eager_rdma_remote.tokens),
mca_btl_udapl_component.udapl_eager_rdma_num);
break;
}
case MCA_BTL_UDAPL_CONTROL_RDMA_CREDIT:
{
mca_btl_udapl_eager_rdma_credit_t* rdma_credit =
frag->segment.seg_addr.pval;
/* don't return credits used for rdma credit control message */
OPAL_THREAD_ADD32(
&(endpoint->endpoint_sr_credits[BTL_UDAPL_EAGER_CONNECTION]),
-1);
OPAL_THREAD_ADD32(&(endpoint->endpoint_eager_rdma_remote.tokens),
rdma_credit->credits);
break;
}
case MCA_BTL_UDAPL_CONTROL_SR_CREDIT:
{
mca_btl_udapl_sr_credit_t* sr_credit =
frag->segment.seg_addr.pval;
/* don't return credits used for sr credit control message */
OPAL_THREAD_ADD32(
&(endpoint->endpoint_sr_credits[sr_credit->connection]), -1);
OPAL_THREAD_ADD32(
&(endpoint->endpoint_sr_tokens[sr_credit->connection]),
sr_credit->credits);
break;
}
default:
BTL_ERROR(("ERROR: Unknown contrl message type received by BTL"));
break;
}
}
/*
* Modify the list of dat entry pointers to include only those entries
* which it is desired to attempt dat_ia_open on.
*
* @param num_info_entries (IN/OUT) Number of entries in datinfo list
* @param datinfo (IN/OUT) List of pointers to dat registry entries
*/
static int mca_btl_udapl_modify_ia_list(DAT_COUNT *num_info_entries,
DAT_PROVIDER_INFO* datinfo)
{
int i,j,k,found;
DAT_PROVIDER_INFO* tmp_datinfo = NULL;
DAT_COUNT tmp_num_entries = 0;
tmp_datinfo = malloc((*num_info_entries) * sizeof(DAT_PROVIDER_INFO));
if(NULL == tmp_datinfo) {
return OMPI_ERR_OUT_OF_RESOURCE;
}
for (i = 0; i < *num_info_entries; i++) {
j = 0;
found = 0;
/* search for datinfo entry on the if_list list */
while (mca_btl_udapl_component.if_list[j]) {
if (0 == strcmp(datinfo[i].ia_name,
mca_btl_udapl_component.if_list[j])) {
found = 1;
/* remove from if_list */
k = opal_argv_count(mca_btl_udapl_component.if_list);
opal_argv_delete(&k, &(mca_btl_udapl_component.if_list),
j, 1);
break;
}
j++;
}
if (found) {
if (NULL != mca_btl_udapl_component.if_include_list) {
/* explicitly include */
tmp_datinfo[tmp_num_entries] = datinfo[i];
tmp_num_entries++;
}
/* if this is if_exclude case and match found do nothing */
} else {
/* if this is if_include case and match not found do nothing */
if (NULL != mca_btl_udapl_component.if_exclude_list) {
/* not found for exclude case so actually include here */
tmp_datinfo[tmp_num_entries] = datinfo[i];
tmp_num_entries++;
}
}
}
/* set new values */
*num_info_entries = tmp_num_entries;
for (j = 0; j < *num_info_entries; j++) {
datinfo[j] = tmp_datinfo[j];
}
/* if if_list not NULL, either not found or user error */
if (opal_argv_count(mca_btl_udapl_component.if_list)) {
char *str = opal_argv_join(mca_btl_udapl_component.if_list, ',');
BTL_UDAPL_VERBOSE_HELP(VERBOSE_SHOW_HELP,
("help-mpi-btl-udapl.txt", "nonexistent entry",
true, orte_process_info.nodename,
((NULL != mca_btl_udapl_component.if_include) ?
"in" : "ex"), str));
free(str);
}
free(tmp_datinfo);
return OMPI_SUCCESS;
}
/*
* Initialize the uDAPL component,
* check how many interfaces are available and create a btl module for each.
*/
mca_btl_base_module_t **
mca_btl_udapl_component_init (int *num_btl_modules,
bool enable_progress_threads,
bool enable_mpi_threads)
{
DAT_PROVIDER_INFO* datinfo;
DAT_PROVIDER_INFO** datinfoptr;
mca_btl_base_module_t **btls;
mca_btl_udapl_module_t *btl;
DAT_COUNT num_ias;
int32_t i;
/* Currently refuse to run if MPI_THREAD_MULTIPLE is enabled */
if (ompi_mpi_thread_multiple && !mca_btl_base_thread_multiple_override) {
mca_btl_udapl_component.udapl_num_btls = 0;
mca_btl_udapl_modex_send();
return NULL;
}
/* parse the include and exclude lists, checking for errors */
mca_btl_udapl_component.if_include_list =
mca_btl_udapl_component.if_exclude_list =
mca_btl_udapl_component.if_list = NULL;
if (NULL != mca_btl_udapl_component.if_include &&
NULL != mca_btl_udapl_component.if_exclude) {
BTL_UDAPL_VERBOSE_HELP(VERBOSE_SHOW_HELP, ("help-mpi-btl-udapl.txt",
"specified include and exclude", true,
mca_btl_udapl_component.if_include,
mca_btl_udapl_component.if_exclude));
mca_btl_udapl_component.udapl_num_btls = 0;
mca_btl_udapl_modex_send();
return NULL;
} else if (NULL != mca_btl_udapl_component.if_include) {
mca_btl_udapl_component.if_include_list =
opal_argv_split(mca_btl_udapl_component.if_include, ',');
mca_btl_udapl_component.if_list =
opal_argv_copy(mca_btl_udapl_component.if_include_list);
} else if (NULL != mca_btl_udapl_component.if_exclude) {
mca_btl_udapl_component.if_exclude_list =
opal_argv_split(mca_btl_udapl_component.if_exclude, ',');
mca_btl_udapl_component.if_list =
opal_argv_copy(mca_btl_udapl_component.if_exclude_list);
}
/* enumerate uDAPL interfaces */
/* Have to do weird pointer stuff to make uDAPL happy -
just an array of DAT_PROVIDER_INFO isn't good enough. */
datinfo = malloc(sizeof(DAT_PROVIDER_INFO) *
mca_btl_udapl_component.udapl_max_btls);
datinfoptr = malloc(sizeof(DAT_PROVIDER_INFO*) *
mca_btl_udapl_component.udapl_max_btls);
if(NULL == datinfo || NULL == datinfoptr) {
return NULL;
}
for(i = 0; i < (int32_t)mca_btl_udapl_component.udapl_max_btls; i++) {
datinfoptr[i] = &datinfo[i];
}
if(DAT_SUCCESS != dat_registry_list_providers(
mca_btl_udapl_component.udapl_max_btls,
(DAT_COUNT*)&num_ias, datinfoptr)) {
free(datinfo);
free(datinfoptr);
return NULL;
}
free(datinfoptr);
/* modify list of IA's to be used when if_in[ex]clude set */
if (NULL != mca_btl_udapl_component.if_list) {
mca_btl_udapl_modify_ia_list(&num_ias, datinfo);
}
/* allocate space for the each possible BTL */
mca_btl_udapl_component.udapl_btls = (mca_btl_udapl_module_t **)
malloc(num_ias * sizeof(mca_btl_udapl_module_t *));
if(NULL == mca_btl_udapl_component.udapl_btls) {
free(datinfo);
return NULL;
}
/* create a BTL module for each interface */
for(mca_btl_udapl_component.udapl_num_btls = i = 0; i < num_ias; i++) {
btl = malloc(sizeof(mca_btl_udapl_module_t));
if(NULL == btl) {
free(datinfo);
free(mca_btl_udapl_component.udapl_btls);
return NULL;
}
/* copy default values into the new BTL */
memcpy(btl, &mca_btl_udapl_module, sizeof(mca_btl_udapl_module_t));
/* initialize this BTL */
/* TODO - make use of the thread-safety info in datinfo also */
if(OMPI_SUCCESS != mca_btl_udapl_init(datinfo[i].ia_name, btl)) {
free(btl);
continue;
}
/* register internal control message callback */
mca_btl_base_active_message_trigger[MCA_BTL_TAG_UDAPL].cbfunc = mca_btl_udapl_receive_control;
mca_btl_base_active_message_trigger[MCA_BTL_TAG_UDAPL].cbdata = NULL;
/* successful btl creation */
mca_btl_udapl_component.udapl_btls[mca_btl_udapl_component.udapl_num_btls] = btl;
if(++mca_btl_udapl_component.udapl_num_btls >=
mca_btl_udapl_component.udapl_max_btls) {
break;
}
}
/* finished with datinfo */
free(datinfo);
/* Make sure we have some interfaces */
if(0 == mca_btl_udapl_component.udapl_num_btls) {
mca_btl_base_error_no_nics("uDAPL", "NIC");
free(mca_btl_udapl_component.udapl_btls);
return NULL;
}
/* publish uDAPL parameters with the MCA framework */
if (OMPI_SUCCESS != mca_btl_udapl_modex_send()) {
free(mca_btl_udapl_component.udapl_btls);
return NULL;
}
/* Post OOB receive */
mca_btl_udapl_endpoint_post_oob_recv();
/* return array of BTLs */
btls = (mca_btl_base_module_t**) malloc(sizeof(mca_btl_base_module_t *) *
mca_btl_udapl_component.udapl_num_btls);
if (NULL == btls) {
free(mca_btl_udapl_component.udapl_btls);
return NULL;
}
memcpy(btls, mca_btl_udapl_component.udapl_btls,
mca_btl_udapl_component.udapl_num_btls *
sizeof(mca_btl_udapl_module_t *));
*num_btl_modules = mca_btl_udapl_component.udapl_num_btls;
return btls;
}
static int mca_btl_udapl_accept_connect(mca_btl_udapl_module_t* btl,
DAT_CR_HANDLE cr_handle)
{
DAT_EP_HANDLE ep;
int rc;
mca_btl_base_endpoint_t* proc_ep;
mca_btl_udapl_addr_t priv_data_in_addr;
int32_t priv_data_in_conn_type; /* incoming endpoint type */
if (mca_btl_udapl_component.udapl_conn_priv_data) {
DAT_CR_PARAM cr_param;
/* query the connection request for incoming private data */
rc = dat_cr_query(cr_handle,
DAT_CR_FIELD_ALL,
&cr_param);
if (rc != DAT_SUCCESS) {
char* major;
char* minor;
dat_strerror(rc, (const char**)&major,
(const char**)&minor);
BTL_ERROR(("ERROR: %s %s %s\n", "dat_cr_query",
major, minor));
return OMPI_ERROR;
}
/* retrieve data from connection request event;
* cr_param contains remote_port_qual but we need to
* match on the psp port and address of remote
* so we get this from the private data.
*/
memcpy(&priv_data_in_addr,
(mca_btl_udapl_addr_t *)cr_param.private_data,
sizeof(mca_btl_udapl_addr_t));
priv_data_in_conn_type = *(int32_t *)
((char *)cr_param.private_data + sizeof(mca_btl_udapl_addr_t));
}
/* create the endpoint for the incoming connection */
rc = mca_btl_udapl_endpoint_create(btl, &ep);
if(OMPI_SUCCESS != rc) {
BTL_ERROR(("ERROR: mca_btl_udapl_endpoint_create"));
return OMPI_ERROR;
}
/* cr_param no longer valid once dat_cr_accept called */
rc = dat_cr_accept(cr_handle, ep, 0, NULL);
if(DAT_SUCCESS != rc) {
char* major;
char* minor;
dat_strerror(rc, (const char**)&major,
(const char**)&minor);
BTL_ERROR(("ERROR: %s %s %s\n", "dat_cr_accept",
major, minor));
return OMPI_ERROR;
}
if (mca_btl_udapl_component.udapl_conn_priv_data) {
/* With accept now in process find a home for the DAT ep by
* matching against the private data that came in on the
* connection request event
*/
/* find the endpoint which matches the address in data received */
proc_ep =
mca_btl_udapl_find_endpoint_address_match(btl, priv_data_in_addr);
if (proc_ep == NULL) {
return OMPI_ERROR;
}
if (BTL_UDAPL_EAGER_CONNECTION == priv_data_in_conn_type) {
proc_ep->endpoint_eager = ep;
} else {
assert(BTL_UDAPL_MAX_CONNECTION == priv_data_in_conn_type);
proc_ep->endpoint_max = ep;
}
}
return OMPI_SUCCESS;
}
static inline int mca_btl_udapl_sendrecv(mca_btl_udapl_module_t* btl,
DAT_EP_HANDLE* endpoint)
{
int rc;
mca_btl_udapl_frag_t* frag;
DAT_DTO_COOKIE cookie;
static int32_t connection_seq = 1;
uint32_t flags = 0;
mca_btl_base_endpoint_t* btl_endpoint = NULL; /* endpoint required by
* mca_btl_udapl_alloc has not
* been created at this point
*/
/* Post a receive to get the peer's address data */
frag = (mca_btl_udapl_frag_t*)
mca_btl_udapl_alloc(
&btl->super,
btl_endpoint,
MCA_BTL_NO_ORDER,
sizeof(mca_btl_udapl_addr_t) +
sizeof(int32_t),
flags);
cookie.as_ptr = frag;
frag->type = MCA_BTL_UDAPL_CONN_RECV;
rc = dat_ep_post_recv(endpoint, 1,
&frag->triplet, cookie, DAT_COMPLETION_DEFAULT_FLAG);
if(DAT_SUCCESS != rc) {
char* major;
char* minor;
dat_strerror(rc, (const char**)&major,
(const char**)&minor);
BTL_ERROR(("ERROR: %s %s %s\n", "dat_ep_post_recv",
major, minor));
return OMPI_ERROR;
}
/* Send our local address data over this EP */
frag = (mca_btl_udapl_frag_t*)
mca_btl_udapl_alloc(
&btl->super,
btl_endpoint,
MCA_BTL_NO_ORDER,
sizeof(mca_btl_udapl_addr_t) +
sizeof(int32_t),
flags);
cookie.as_ptr = frag;
memcpy(frag->segment.seg_addr.pval,
&btl->udapl_addr, sizeof(mca_btl_udapl_addr_t));
memcpy((char *)frag->segment.seg_addr.pval + sizeof(mca_btl_udapl_addr_t),
&connection_seq, sizeof(int32_t));
connection_seq++;
frag->type = MCA_BTL_UDAPL_CONN_SEND;
rc = dat_ep_post_send(endpoint, 1,
&frag->triplet, cookie, DAT_COMPLETION_DEFAULT_FLAG);
if(DAT_SUCCESS != rc) {
char* major;
char* minor;
dat_strerror(rc, (const char**)&major,
(const char**)&minor);
BTL_ERROR(("ERROR: %s %s %s\n", "dat_ep_post_send",
major, minor));
return OMPI_ERROR;
}
return OMPI_SUCCESS;
}
static inline int mca_btl_udapl_frag_progress_one(
mca_btl_udapl_module_t* udapl_btl,
mca_btl_udapl_frag_t* frag)
{
int rc;
switch(frag->type) {
case MCA_BTL_UDAPL_SEND:
rc = mca_btl_udapl_endpoint_send(frag->endpoint, frag);
break;
case MCA_BTL_UDAPL_PUT:
rc = mca_btl_udapl_put(&udapl_btl->super,
frag->endpoint,
&frag->base);
break;
default:
rc = OMPI_ERROR;
BTL_ERROR(("Error : Progressing pending operation, invalid type %d\n",
frag->type));
break;
}
return rc;
}
void mca_btl_udapl_frag_progress_pending(mca_btl_udapl_module_t* udapl_btl,
mca_btl_base_endpoint_t* endpoint,
const int connection)
{
int len;
int i;
int token_avail;
mca_btl_udapl_frag_t* frag;
if (BTL_UDAPL_EAGER_CONNECTION == connection) {
len = opal_list_get_size(&endpoint->endpoint_eager_frags);
/* progress eager frag queue as needed */
BTL_UDAPL_TOKEN_AVAIL(endpoint, connection, token_avail);
for(i = 0; i < len && token_avail > 0; i++) {
OPAL_THREAD_LOCK(&endpoint->endpoint_lock);
frag = (mca_btl_udapl_frag_t*)opal_list_remove_first(&(endpoint->endpoint_eager_frags));
OPAL_THREAD_UNLOCK(&endpoint->endpoint_lock);
if(NULL == frag) {
return;
}
if(mca_btl_udapl_frag_progress_one(udapl_btl, frag) !=
OMPI_SUCCESS) {
BTL_ERROR(("ERROR: Not able to progress on connection(%d)\n",
BTL_UDAPL_EAGER_CONNECTION));
return;
}
BTL_UDAPL_TOKEN_AVAIL(endpoint, connection, token_avail);
}
} else if (BTL_UDAPL_MAX_CONNECTION == connection) {
len = opal_list_get_size(&endpoint->endpoint_max_frags);
BTL_UDAPL_TOKEN_AVAIL(endpoint, connection, token_avail);
/* progress max frag queue as needed */
for(i = 0; i < len && token_avail > 0; i++) {
OPAL_THREAD_LOCK(&endpoint->endpoint_lock);
frag = (mca_btl_udapl_frag_t*)opal_list_remove_first(&(endpoint->endpoint_max_frags));
OPAL_THREAD_UNLOCK(&endpoint->endpoint_lock);
if(NULL == frag) {
return;
}
if(mca_btl_udapl_frag_progress_one(udapl_btl, frag) !=
OMPI_SUCCESS) {
BTL_ERROR(("ERROR: Not able to progress on connection(%d)\n",
BTL_UDAPL_MAX_CONNECTION));
return;
}
BTL_UDAPL_TOKEN_AVAIL(endpoint, connection, token_avail);
}
} else {
BTL_ERROR(("ERROR: Can not progress pending fragment on unknown connection\n"));
}
return;
}
/*
* uDAPL component progress.
*/
int mca_btl_udapl_component_progress()
{
mca_btl_udapl_module_t* btl;
static int32_t inprogress = 0;
DAT_EVENT event;
size_t i;
int32_t j, rdma_ep_count;
int count = 0, btl_ownership;
mca_btl_udapl_frag_t* frag;
mca_btl_base_endpoint_t* endpoint;
/* prevent deadlock - only one thread should be 'progressing' at a time */
if(OPAL_THREAD_ADD32(&inprogress, 1) > 1) {
OPAL_THREAD_ADD32(&inprogress, -1);
return OMPI_SUCCESS;
}
/* check for work to do on each uDAPL btl */
OPAL_THREAD_LOCK(&mca_btl_udapl_component.udapl_lock);
for(i = 0; i < mca_btl_udapl_component.udapl_num_btls; i++) {
btl = mca_btl_udapl_component.udapl_btls[i];
/* Check DTO EVD */
while(DAT_SUCCESS ==
dat_evd_dequeue(btl->udapl_evd_dto, &event)) {
DAT_DTO_COMPLETION_EVENT_DATA* dto;
switch(event.event_number) {
case DAT_DTO_COMPLETION_EVENT:
dto = &event.event_data.dto_completion_event_data;
frag = dto->user_cookie.as_ptr;
/* Was the DTO successful? */
if(DAT_DTO_SUCCESS != dto->status) {
if (DAT_DTO_ERR_FLUSHED == dto->status) {
BTL_UDAPL_VERBOSE_OUTPUT(VERBOSE_INFORM,
("DAT_DTO_ERR_FLUSHED: probably OK if occurs during MPI_Finalize().\n"));
} else {
BTL_UDAPL_VERBOSE_OUTPUT(VERBOSE_CRITICAL,
("ERROR: DAT_DTO_COMPLETION_EVENT: %d %d %lu %p.\n",
dto->status, frag->type,
(unsigned long)frag->size, dto->ep_handle));
}
return OMPI_ERROR;
}
endpoint = frag->endpoint;
btl_ownership = (frag->base.des_flags & MCA_BTL_DES_FLAGS_BTL_OWNERSHIP);
switch(frag->type) {
case MCA_BTL_UDAPL_RDMA_WRITE:
{
assert(frag->base.des_src == &frag->segment);
assert(frag->base.des_src_cnt == 1);
assert(frag->base.des_dst == NULL);
assert(frag->base.des_dst_cnt == 0);
assert(frag->type == MCA_BTL_UDAPL_RDMA_WRITE);
frag->base.des_cbfunc(&btl->super, endpoint,
&frag->base, OMPI_SUCCESS);
if( btl_ownership ) {
mca_btl_udapl_free(&btl->super,
&frag->base);
}
OPAL_THREAD_ADD32(&(endpoint->endpoint_lwqe_tokens[BTL_UDAPL_EAGER_CONNECTION]), 1);
mca_btl_udapl_frag_progress_pending(btl,
endpoint, BTL_UDAPL_EAGER_CONNECTION);
break;
}
case MCA_BTL_UDAPL_SEND:
{
int connection = BTL_UDAPL_EAGER_CONNECTION;
assert(frag->base.des_src == &frag->segment);
assert(frag->base.des_src_cnt == 1);
assert(frag->base.des_dst == NULL);
assert(frag->base.des_dst_cnt == 0);
assert(frag->type == MCA_BTL_UDAPL_SEND);
if(frag->size !=
mca_btl_udapl_component.udapl_eager_frag_size) {
assert(frag->size ==
mca_btl_udapl_component.udapl_max_frag_size);
connection = BTL_UDAPL_MAX_CONNECTION;
}
frag->base.des_cbfunc(&btl->super, endpoint,
&frag->base, OMPI_SUCCESS);
if( btl_ownership ) {
mca_btl_udapl_free(&btl->super,
&frag->base);
}
OPAL_THREAD_ADD32(&(endpoint->endpoint_lwqe_tokens[connection]), 1);
mca_btl_udapl_frag_progress_pending(btl,
endpoint, connection);
break;
}
case MCA_BTL_UDAPL_RECV:
{
mca_btl_active_message_callback_t* reg;
int cntrl_msg = -1;
assert(frag->base.des_dst == &frag->segment);
assert(frag->base.des_dst_cnt == 1);
assert(frag->base.des_src == NULL);
assert(frag->base.des_src_cnt == 0);
assert(frag->type == MCA_BTL_UDAPL_RECV);
assert(frag->triplet.virtual_address ==
(DAT_VADDR)(uintptr_t)frag->segment.seg_addr.pval);
assert(frag->triplet.segment_length == frag->size);
assert(frag->btl == btl);
/* setup frag ftr location and do callback */
frag->segment.seg_len = dto->transfered_length -
sizeof(mca_btl_udapl_footer_t);
frag->ftr = (mca_btl_udapl_footer_t *)
((char *)frag->segment.seg_addr.pval +
frag->segment.seg_len);
cntrl_msg = frag->ftr->tag;
reg = mca_btl_base_active_message_trigger + frag->ftr->tag;
OPAL_THREAD_UNLOCK(&mca_btl_udapl_component.udapl_lock);
reg->cbfunc(&btl->super,
frag->ftr->tag, &frag->base, reg->cbdata);
OPAL_THREAD_LOCK(&mca_btl_udapl_component.udapl_lock);
/* Repost the frag */
frag->ftr = frag->segment.seg_addr.pval;
frag->segment.seg_len =
(frag->size - sizeof(mca_btl_udapl_footer_t) -
sizeof(mca_btl_udapl_rdma_footer_t));
frag->base.des_flags = 0;
if(frag->size ==
mca_btl_udapl_component.udapl_eager_frag_size) {
OPAL_THREAD_ADD32(&(frag->endpoint->endpoint_sr_credits[BTL_UDAPL_EAGER_CONNECTION]), 1);
dat_ep_post_recv(frag->endpoint->endpoint_eager,
1, &frag->triplet, dto->user_cookie,
DAT_COMPLETION_DEFAULT_FLAG);
if (frag->endpoint->endpoint_sr_credits[BTL_UDAPL_EAGER_CONNECTION] >=
mca_btl_udapl_component.udapl_sr_win) {
mca_btl_udapl_endpoint_send_sr_credits(frag->endpoint,
BTL_UDAPL_EAGER_CONNECTION);
}
if (MCA_BTL_TAG_UDAPL == cntrl_msg) {
mca_btl_udapl_frag_progress_pending(btl,
frag->endpoint,
BTL_UDAPL_EAGER_CONNECTION);
}
} else {
assert(frag->size ==
mca_btl_udapl_component.udapl_max_frag_size);
OPAL_THREAD_ADD32(&(frag->endpoint->endpoint_sr_credits[BTL_UDAPL_MAX_CONNECTION]), 1);
dat_ep_post_recv(frag->endpoint->endpoint_max,
1, &frag->triplet, dto->user_cookie,
DAT_COMPLETION_DEFAULT_FLAG);
if (frag->endpoint->endpoint_sr_credits[BTL_UDAPL_MAX_CONNECTION] >=
mca_btl_udapl_component.udapl_sr_win) {
mca_btl_udapl_endpoint_send_sr_credits(frag->endpoint,
BTL_UDAPL_MAX_CONNECTION);
}
if (MCA_BTL_TAG_UDAPL == cntrl_msg) {
mca_btl_udapl_frag_progress_pending(btl,
frag->endpoint,
BTL_UDAPL_MAX_CONNECTION);
}
}
break;
}
case MCA_BTL_UDAPL_PUT:
{
assert(frag->base.des_src == &frag->segment);
assert(frag->base.des_src_cnt == 1);
assert(frag->base.des_dst_cnt == 1);
assert(frag->type == MCA_BTL_UDAPL_PUT);
frag->base.des_cbfunc(&btl->super, endpoint,
&frag->base, OMPI_SUCCESS);
if( btl_ownership ) {
mca_btl_udapl_free(&btl->super,
&frag->base);
}
OPAL_THREAD_ADD32(&(endpoint->endpoint_lwqe_tokens[BTL_UDAPL_MAX_CONNECTION]), 1);
OPAL_THREAD_ADD32(&(endpoint->endpoint_sr_tokens[BTL_UDAPL_MAX_CONNECTION]), 1);
mca_btl_udapl_frag_progress_pending(btl,
endpoint, BTL_UDAPL_MAX_CONNECTION);
break;
}
case MCA_BTL_UDAPL_CONN_RECV:
mca_btl_udapl_endpoint_finish_connect(btl,
frag->segment.seg_addr.pval,
(int32_t *)((char *)frag->segment.seg_addr.pval +
sizeof(mca_btl_udapl_addr_t)),
event.event_data.connect_event_data.ep_handle);
/* No break - fall through to free */
case MCA_BTL_UDAPL_CONN_SEND:
frag->segment.seg_len =
mca_btl_udapl_module.super.btl_eager_limit;
mca_btl_udapl_free(&btl->super, &frag->base);
break;
default:
BTL_UDAPL_VERBOSE_OUTPUT(VERBOSE_DIAGNOSE,
("WARNING: unknown frag type: %d\n",
frag->type));
}
count++;
break;
default:
BTL_UDAPL_VERBOSE_OUTPUT(VERBOSE_DIAGNOSE,
("WARNING: DTO event: %s (%d)\n",
mca_btl_udapl_dat_event_to_string(event.event_number),
event.event_number));
}
}
/* Check connection EVD */
while((btl->udapl_connect_inprogress > 0) && (DAT_SUCCESS ==
dat_evd_dequeue(btl->udapl_evd_conn, &event))) {
switch(event.event_number) {
case DAT_CONNECTION_REQUEST_EVENT:
/* Accept a new connection */
mca_btl_udapl_accept_connect(btl,
event.event_data.cr_arrival_event_data.cr_handle);
count++;
break;
case DAT_CONNECTION_EVENT_ESTABLISHED:
/* Both the client and server side of a connection generate
this event */
if (mca_btl_udapl_component.udapl_conn_priv_data) {
/* private data is only valid at this point if this
* event is from a dat_ep_connect call, not an accept
*/
mca_btl_udapl_endpoint_pd_established_conn(btl,
event.event_data.connect_event_data.ep_handle);
} else {
/* explicitly exchange process data */
mca_btl_udapl_sendrecv(btl,
event.event_data.connect_event_data.ep_handle);
}
count++;
break;
case DAT_CONNECTION_EVENT_PEER_REJECTED:
case DAT_CONNECTION_EVENT_NON_PEER_REJECTED:
case DAT_CONNECTION_EVENT_ACCEPT_COMPLETION_ERROR:
case DAT_CONNECTION_EVENT_DISCONNECTED:
case DAT_CONNECTION_EVENT_BROKEN:
case DAT_CONNECTION_EVENT_TIMED_OUT:
/* handle this case specially? if we have finite timeout,
we might want to try connecting again here. */
case DAT_CONNECTION_EVENT_UNREACHABLE:
/* Need to set the BTL endpoint to MCA_BTL_UDAPL_FAILED
See dat_ep_connect documentation pdf pg 198 */
BTL_UDAPL_VERBOSE_OUTPUT(VERBOSE_CRITICAL,
("WARNING: connection event not handled : %s (%d)\n",
mca_btl_udapl_dat_event_to_string(event.event_number),
event.event_number));
break;
default:
BTL_ERROR(("ERROR: connection event : %s (%d)",
mca_btl_udapl_dat_event_to_string(event.event_number),
event.event_number));
}
}
/* Check async EVD */
if (btl->udapl_async_events == mca_btl_udapl_component.udapl_async_events) {
btl->udapl_async_events = 0;
while(DAT_SUCCESS ==
dat_evd_dequeue(btl->udapl_evd_async, &event)) {
switch(event.event_number) {
case DAT_ASYNC_ERROR_EVD_OVERFLOW:
case DAT_ASYNC_ERROR_IA_CATASTROPHIC:
case DAT_ASYNC_ERROR_EP_BROKEN:
case DAT_ASYNC_ERROR_TIMED_OUT:
case DAT_ASYNC_ERROR_PROVIDER_INTERNAL_ERROR:
BTL_UDAPL_VERBOSE_OUTPUT(VERBOSE_CRITICAL,
("WARNING: async event ignored : %s (%d)",
mca_btl_udapl_dat_event_to_string(event.event_number),
event.event_number));
break;
default:
BTL_UDAPL_VERBOSE_OUTPUT(VERBOSE_CRITICAL,
("WARNING: %s (%d)\n",
mca_btl_udapl_dat_event_to_string(event.event_number),
event.event_number));
}
}
} else {
btl->udapl_async_events++;
}
/*
* Check eager rdma segments
*/
/* find the number of endpoints with rdma buffers */
rdma_ep_count = btl->udapl_eager_rdma_endpoint_count;
for (j = 0; j < rdma_ep_count; j++) {
mca_btl_udapl_endpoint_t* endpoint;
mca_btl_udapl_frag_t *local_rdma_frag;
endpoint =
opal_pointer_array_get_item(btl->udapl_eager_rdma_endpoints, j);
OPAL_THREAD_LOCK(&endpoint->endpoint_eager_rdma_local.lock);
local_rdma_frag =
MCA_BTL_UDAPL_GET_LOCAL_RDMA_FRAG(endpoint,
endpoint->endpoint_eager_rdma_local.head);
if (local_rdma_frag->rdma_ftr->active == 1) {
int pad = 0;
mca_btl_active_message_callback_t* reg;
MCA_BTL_UDAPL_RDMA_NEXT_INDEX(endpoint->endpoint_eager_rdma_local.head);
OPAL_THREAD_UNLOCK(&endpoint->endpoint_eager_rdma_local.lock);
/* compute pad as needed */
MCA_BTL_UDAPL_FRAG_CALC_ALIGNMENT_PAD(pad,
(local_rdma_frag->rdma_ftr->size +
sizeof(mca_btl_udapl_footer_t)));
/* set fragment information */
local_rdma_frag->ftr = (mca_btl_udapl_footer_t *)
((char *)local_rdma_frag->rdma_ftr -
pad -
sizeof(mca_btl_udapl_footer_t));
local_rdma_frag->segment.seg_len =
local_rdma_frag->rdma_ftr->size;
local_rdma_frag->segment.seg_addr.pval = (unsigned char *)
((char *)local_rdma_frag->ftr -
local_rdma_frag->segment.seg_len);
/* trigger callback */
reg = mca_btl_base_active_message_trigger + local_rdma_frag->ftr->tag;
reg->cbfunc(&btl->super,
local_rdma_frag->ftr->tag, &local_rdma_frag->base, reg->cbdata);
/* repost */
local_rdma_frag->rdma_ftr->active = 0;
local_rdma_frag->segment.seg_len =
mca_btl_udapl_module.super.btl_eager_limit;
local_rdma_frag->base.des_flags = 0;
/* increment local rdma credits */
OPAL_THREAD_ADD32(&(endpoint->endpoint_eager_rdma_local.credits),
1);
if (endpoint->endpoint_eager_rdma_local.credits >=
mca_btl_udapl_component.udapl_eager_rdma_win) {
mca_btl_udapl_endpoint_send_eager_rdma_credits(endpoint);
}
count++;
} else {
OPAL_THREAD_UNLOCK(&endpoint->endpoint_eager_rdma_local.lock);
}
} /* end of rdma_count loop */
}
/* unlock and return */
OPAL_THREAD_UNLOCK(&mca_btl_udapl_component.udapl_lock);
OPAL_THREAD_ADD32(&inprogress, -1);
return count;
}
Показать git blame Копировать постоянную ссылку