388 строки
12 KiB
C
388 строки
12 KiB
C
/*
|
|
* Copyright (c) 2004-2005 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) 2007 Sun Microsystems, Inc. All rights reserved.
|
|
* $COPYRIGHT$
|
|
*
|
|
* Additional copyrights may follow
|
|
*
|
|
* $HEADER$
|
|
*/
|
|
|
|
#include "ompi_config.h"
|
|
#include "opal/threads/mutex.h"
|
|
#include "opal/threads/condition.h"
|
|
#include "ompi/datatype/ompi_datatype.h"
|
|
#include "opal/mca/allocator/base/base.h"
|
|
#include "opal/mca/allocator/allocator.h"
|
|
#include "ompi/mca/pml/pml.h"
|
|
#include "ompi/mca/pml/base/pml_base_request.h"
|
|
#include "ompi/mca/pml/base/pml_base_sendreq.h"
|
|
#include "ompi/mca/pml/base/pml_base_bsend.h"
|
|
#include "opal/mca/mpool/mpool.h"
|
|
|
|
#ifdef HAVE_UNISTD_H
|
|
#include <unistd.h>
|
|
#endif /* HAVE_UNISTD_H */
|
|
|
|
static opal_mutex_t mca_pml_bsend_mutex; /* lock for thread safety */
|
|
static opal_condition_t mca_pml_bsend_condition; /* condition variable to block on detach */
|
|
static mca_allocator_base_component_t* mca_pml_bsend_allocator_component;
|
|
static mca_allocator_base_module_t* mca_pml_bsend_allocator; /* sub-allocator to manage users buffer */
|
|
static size_t mca_pml_bsend_usersize; /* user provided buffer size */
|
|
static unsigned char *mca_pml_bsend_userbase; /* user provided buffer base */
|
|
static unsigned char *mca_pml_bsend_base; /* adjusted base of user buffer */
|
|
static unsigned char *mca_pml_bsend_addr; /* current offset into user buffer */
|
|
static size_t mca_pml_bsend_size; /* adjusted size of user buffer */
|
|
static size_t mca_pml_bsend_count; /* number of outstanding requests */
|
|
static size_t mca_pml_bsend_pagesz; /* mmap page size */
|
|
static int mca_pml_bsend_pagebits; /* number of bits in pagesz */
|
|
static int32_t mca_pml_bsend_init = 0;
|
|
|
|
/* defined in pml_base_open.c */
|
|
extern char *ompi_pml_base_bsend_allocator_name;
|
|
|
|
/*
|
|
* Routine to return pages to sub-allocator as needed
|
|
*/
|
|
static void* mca_pml_bsend_alloc_segment(
|
|
struct mca_mpool_base_module_t* module,
|
|
size_t* size_inout,
|
|
mca_mpool_base_registration_t** registration)
|
|
{
|
|
void *addr;
|
|
size_t size = *size_inout;
|
|
if(mca_pml_bsend_addr + size > mca_pml_bsend_base + mca_pml_bsend_size) {
|
|
return NULL;
|
|
}
|
|
/* allocate all that is left */
|
|
size = mca_pml_bsend_size - (mca_pml_bsend_addr - mca_pml_bsend_base);
|
|
addr = mca_pml_bsend_addr;
|
|
mca_pml_bsend_addr += size;
|
|
*size_inout = size;
|
|
if (NULL != registration) *registration = NULL;
|
|
return addr;
|
|
}
|
|
|
|
/*
|
|
* One time initialization at startup
|
|
*/
|
|
int mca_pml_base_bsend_init(bool thread_safe)
|
|
{
|
|
size_t tmp;
|
|
|
|
if(OPAL_THREAD_ADD32(&mca_pml_bsend_init, 1) > 1)
|
|
return OMPI_SUCCESS;
|
|
|
|
/* initialize static objects */
|
|
OBJ_CONSTRUCT(&mca_pml_bsend_mutex, opal_mutex_t);
|
|
OBJ_CONSTRUCT(&mca_pml_bsend_condition, opal_condition_t);
|
|
|
|
/* lookup name of the allocator to use for buffered sends */
|
|
if(NULL == (mca_pml_bsend_allocator_component = mca_allocator_component_lookup(ompi_pml_base_bsend_allocator_name))) {
|
|
return OMPI_ERR_BUFFER;
|
|
}
|
|
|
|
/* determine page size */
|
|
tmp = mca_pml_bsend_pagesz = sysconf(_SC_PAGESIZE);
|
|
mca_pml_bsend_pagebits = 0;
|
|
while( tmp != 0 ) {
|
|
tmp >>= 1;
|
|
mca_pml_bsend_pagebits++;
|
|
}
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
|
|
/*
|
|
* One-time cleanup at shutdown - release any resources.
|
|
*/
|
|
int mca_pml_base_bsend_fini(void)
|
|
{
|
|
if(OPAL_THREAD_ADD32(&mca_pml_bsend_init,-1) > 0)
|
|
return OMPI_SUCCESS;
|
|
|
|
if(NULL != mca_pml_bsend_allocator)
|
|
mca_pml_bsend_allocator->alc_finalize(mca_pml_bsend_allocator);
|
|
mca_pml_bsend_allocator = NULL;
|
|
|
|
OBJ_DESTRUCT(&mca_pml_bsend_condition);
|
|
OBJ_DESTRUCT(&mca_pml_bsend_mutex);
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
|
|
/*
|
|
* User-level call to attach buffer.
|
|
*/
|
|
int mca_pml_base_bsend_attach(void* addr, int size)
|
|
{
|
|
int align;
|
|
|
|
bool thread_safe = ompi_mpi_thread_multiple;
|
|
if(NULL == addr || size <= 0) {
|
|
return OMPI_ERR_BUFFER;
|
|
}
|
|
|
|
/* check for buffer already attached */
|
|
OPAL_THREAD_LOCK(&mca_pml_bsend_mutex);
|
|
if(NULL != mca_pml_bsend_allocator) {
|
|
OPAL_THREAD_UNLOCK(&mca_pml_bsend_mutex);
|
|
return OMPI_ERR_BUFFER;
|
|
}
|
|
|
|
/* try to create an instance of the allocator - to determine thread safety level */
|
|
mca_pml_bsend_allocator = mca_pml_bsend_allocator_component->allocator_init(thread_safe, mca_pml_bsend_alloc_segment, NULL, NULL);
|
|
if(NULL == mca_pml_bsend_allocator) {
|
|
OPAL_THREAD_UNLOCK(&mca_pml_bsend_mutex);
|
|
return OMPI_ERR_BUFFER;
|
|
}
|
|
|
|
/*
|
|
* Save away what the user handed in. This is done in case the
|
|
* base and size are modified for alignment issues.
|
|
*/
|
|
mca_pml_bsend_userbase = (unsigned char*)addr;
|
|
mca_pml_bsend_usersize = size;
|
|
/*
|
|
* Align to pointer boundaries. The bsend overhead is large enough
|
|
* to account for this. Compute any alignment that needs to be done.
|
|
*/
|
|
align = sizeof(void *) - ((size_t)addr & (sizeof(void *) - 1));
|
|
|
|
/* setup local variables */
|
|
mca_pml_bsend_base = (unsigned char *)addr + align;
|
|
mca_pml_bsend_addr = (unsigned char *)addr + align;
|
|
mca_pml_bsend_size = size - align;
|
|
mca_pml_bsend_count = 0;
|
|
OPAL_THREAD_UNLOCK(&mca_pml_bsend_mutex);
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
/*
|
|
* User-level call to detach buffer
|
|
*/
|
|
int mca_pml_base_bsend_detach(void* addr, int* size)
|
|
{
|
|
OPAL_THREAD_LOCK(&mca_pml_bsend_mutex);
|
|
|
|
/* is buffer attached */
|
|
if(NULL == mca_pml_bsend_allocator) {
|
|
OPAL_THREAD_UNLOCK(&mca_pml_bsend_mutex);
|
|
return OMPI_ERR_BUFFER;
|
|
}
|
|
|
|
/* wait on any pending requests */
|
|
while(mca_pml_bsend_count != 0)
|
|
opal_condition_wait(&mca_pml_bsend_condition, &mca_pml_bsend_mutex);
|
|
|
|
/* free resources associated with the allocator */
|
|
mca_pml_bsend_allocator->alc_finalize(mca_pml_bsend_allocator);
|
|
mca_pml_bsend_allocator = NULL;
|
|
|
|
/* return current settings */
|
|
if(NULL != addr)
|
|
*((void**)addr) = mca_pml_bsend_userbase;
|
|
if(NULL != size)
|
|
*size = (int)mca_pml_bsend_usersize;
|
|
|
|
/* reset local variables */
|
|
mca_pml_bsend_userbase = NULL;
|
|
mca_pml_bsend_usersize = 0;
|
|
mca_pml_bsend_base = NULL;
|
|
mca_pml_bsend_addr = NULL;
|
|
mca_pml_bsend_size = 0;
|
|
mca_pml_bsend_count = 0;
|
|
OPAL_THREAD_UNLOCK(&mca_pml_bsend_mutex);
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
|
|
/*
|
|
* pack send buffer into buffer
|
|
*/
|
|
|
|
int mca_pml_base_bsend_request_start(ompi_request_t* request)
|
|
{
|
|
mca_pml_base_send_request_t* sendreq = (mca_pml_base_send_request_t*)request;
|
|
struct iovec iov;
|
|
unsigned int iov_count;
|
|
size_t max_data;
|
|
int rc;
|
|
|
|
if(sendreq->req_bytes_packed > 0) {
|
|
|
|
/* has a buffer been provided */
|
|
OPAL_THREAD_LOCK(&mca_pml_bsend_mutex);
|
|
if(NULL == mca_pml_bsend_addr) {
|
|
sendreq->req_addr = NULL;
|
|
OPAL_THREAD_UNLOCK(&mca_pml_bsend_mutex);
|
|
return OMPI_ERR_BUFFER;
|
|
}
|
|
|
|
/* allocate a buffer to hold packed message */
|
|
sendreq->req_addr = mca_pml_bsend_allocator->alc_alloc(
|
|
mca_pml_bsend_allocator, sendreq->req_bytes_packed, 0, NULL);
|
|
if(NULL == sendreq->req_addr) {
|
|
/* release resources when request is freed */
|
|
sendreq->req_base.req_pml_complete = true;
|
|
OPAL_THREAD_UNLOCK(&mca_pml_bsend_mutex);
|
|
return OMPI_ERR_BUFFER;
|
|
}
|
|
|
|
OPAL_THREAD_UNLOCK(&mca_pml_bsend_mutex);
|
|
|
|
/* The convertor is already initialized in the begining so we just have to
|
|
* pack the data in the newly allocated buffer.
|
|
*/
|
|
iov.iov_base = (IOVBASE_TYPE*)sendreq->req_addr;
|
|
iov.iov_len = sendreq->req_bytes_packed;
|
|
iov_count = 1;
|
|
max_data = iov.iov_len;
|
|
if((rc = opal_convertor_pack( &sendreq->req_base.req_convertor,
|
|
&iov,
|
|
&iov_count,
|
|
&max_data )) < 0) {
|
|
return OMPI_ERROR;
|
|
}
|
|
|
|
/* setup convertor to point to packed buffer (at position zero) */
|
|
opal_convertor_prepare_for_send( &sendreq->req_base.req_convertor, &(ompi_mpi_packed.dt.super),
|
|
max_data, sendreq->req_addr );
|
|
/* increment count of pending requests */
|
|
mca_pml_bsend_count++;
|
|
}
|
|
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
|
|
/*
|
|
* allocate buffer
|
|
*/
|
|
|
|
int mca_pml_base_bsend_request_alloc(ompi_request_t* request)
|
|
{
|
|
mca_pml_base_send_request_t* sendreq = (mca_pml_base_send_request_t*)request;
|
|
|
|
assert( sendreq->req_bytes_packed > 0 );
|
|
|
|
/* has a buffer been provided */
|
|
OPAL_THREAD_LOCK(&mca_pml_bsend_mutex);
|
|
if(NULL == mca_pml_bsend_addr) {
|
|
sendreq->req_addr = NULL;
|
|
OPAL_THREAD_UNLOCK(&mca_pml_bsend_mutex);
|
|
return OMPI_ERR_BUFFER;
|
|
}
|
|
|
|
/* allocate a buffer to hold packed message */
|
|
sendreq->req_addr = mca_pml_bsend_allocator->alc_alloc(
|
|
mca_pml_bsend_allocator, sendreq->req_bytes_packed, 0, NULL);
|
|
if(NULL == sendreq->req_addr) {
|
|
/* release resources when request is freed */
|
|
sendreq->req_base.req_pml_complete = true;
|
|
OPAL_THREAD_UNLOCK(&mca_pml_bsend_mutex);
|
|
/* progress communications, with the hope that more resources
|
|
* will be freed */
|
|
opal_progress();
|
|
return OMPI_ERR_BUFFER;
|
|
}
|
|
|
|
/* increment count of pending requests */
|
|
mca_pml_bsend_count++;
|
|
OPAL_THREAD_UNLOCK(&mca_pml_bsend_mutex);
|
|
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
/*
|
|
* allocate buffer
|
|
*/
|
|
|
|
void* mca_pml_base_bsend_request_alloc_buf( size_t length )
|
|
{
|
|
void* buf = NULL;
|
|
/* has a buffer been provided */
|
|
OPAL_THREAD_LOCK(&mca_pml_bsend_mutex);
|
|
if(NULL == mca_pml_bsend_addr) {
|
|
OPAL_THREAD_UNLOCK(&mca_pml_bsend_mutex);
|
|
return NULL;
|
|
}
|
|
|
|
/* allocate a buffer to hold packed message */
|
|
buf = mca_pml_bsend_allocator->alc_alloc(
|
|
mca_pml_bsend_allocator, length, 0, NULL);
|
|
if(NULL == buf) {
|
|
/* release resources when request is freed */
|
|
OPAL_THREAD_UNLOCK(&mca_pml_bsend_mutex);
|
|
/* progress communications, with the hope that more resources
|
|
* will be freed */
|
|
opal_progress();
|
|
return NULL;
|
|
}
|
|
|
|
/* increment count of pending requests */
|
|
mca_pml_bsend_count++;
|
|
OPAL_THREAD_UNLOCK(&mca_pml_bsend_mutex);
|
|
|
|
return buf;
|
|
}
|
|
|
|
|
|
/*
|
|
* Request completed - free buffer and decrement pending count
|
|
*/
|
|
int mca_pml_base_bsend_request_free(void* addr)
|
|
{
|
|
/* remove from list of pending requests */
|
|
OPAL_THREAD_LOCK(&mca_pml_bsend_mutex);
|
|
|
|
/* free buffer */
|
|
mca_pml_bsend_allocator->alc_free(mca_pml_bsend_allocator, addr);
|
|
|
|
/* decrement count of buffered requests */
|
|
if(--mca_pml_bsend_count == 0)
|
|
opal_condition_signal(&mca_pml_bsend_condition);
|
|
|
|
OPAL_THREAD_UNLOCK(&mca_pml_bsend_mutex);
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
|
|
|
|
/*
|
|
* Request completed - free buffer and decrement pending count
|
|
*/
|
|
int mca_pml_base_bsend_request_fini(ompi_request_t* request)
|
|
{
|
|
mca_pml_base_send_request_t* sendreq = (mca_pml_base_send_request_t*)request;
|
|
if(sendreq->req_bytes_packed == 0 ||
|
|
sendreq->req_addr == NULL ||
|
|
sendreq->req_addr == sendreq->req_base.req_addr)
|
|
return OMPI_SUCCESS;
|
|
|
|
/* remove from list of pending requests */
|
|
OPAL_THREAD_LOCK(&mca_pml_bsend_mutex);
|
|
|
|
/* free buffer */
|
|
mca_pml_bsend_allocator->alc_free(mca_pml_bsend_allocator, sendreq->req_addr);
|
|
sendreq->req_addr = sendreq->req_base.req_addr;
|
|
|
|
/* decrement count of buffered requests */
|
|
if(--mca_pml_bsend_count == 0)
|
|
opal_condition_signal(&mca_pml_bsend_condition);
|
|
|
|
OPAL_THREAD_UNLOCK(&mca_pml_bsend_mutex);
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
|