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

Improve the latency for small and medium messages. The idea is to decrease the

Просмотр исходного кода 
number of recv system call by caching the data. Each endpoint has a buffer
(the size is an MCA parameter) that can be use as a cache. Before each receive
operation this buffer is added at the end of the iovec list. All data that are
not expected by the fragment will go in this cache. If the cache contain data
all subsequent receive will just memcpy the data into the BTL buffers.

The only drawback is that we will spin around the receive_handle until all the
cached data is readed by the PML layer. This limitation come from the fact that
the event library is unable to call us if there is no events on the socket.
Therefore we are unable to keep the data in the cache until the next loop
into the progress engine.

This commit was SVN r8398.
Этот коммит содержится в:
George Bosilca 2005-12-07 00:12:59 +00:00
родитель b04f6bf71b
Коммит 5851b55647
5 изменённых файлов: 168 добавлений и 106 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

9
ompi/mca/btl/tcp/btl_tcp.h
Просмотреть файл

@ -33,7 +33,7 @@
#ifdef HAVE_NETINET_IN_H
#include <netinet/in.h>
#endif
/* Open MPI includes */
#include "opal/event/event.h"
#include "opal/util/output.h"
@ -64,6 +64,7 @@ struct mca_btl_tcp_component_t {
int tcp_free_list_num; /**< initial size of free lists */
int tcp_free_list_max; /**< maximum size of free lists */
int tcp_free_list_inc; /**< number of elements to alloc when growing free lists */
int tcp_endpoint_cache; /**< amount of cache on each endpoint */
opal_hash_table_t tcp_procs; /**< hash table of tcp proc structures */
opal_list_t tcp_events; /**< list of pending tcp events */
opal_mutex_t tcp_lock; /**< lock for accessing module state */
@ -209,7 +210,7 @@ extern int mca_btl_tcp_del_procs(
extern int mca_btl_tcp_send(
struct mca_btl_base_module_t* btl,
struct mca_btl_base_endpoint_t* btl_peer,
struct mca_btl_base_descriptor_t* descriptor,
struct mca_btl_base_descriptor_t* descriptor,
mca_btl_base_tag_t tag
);
@ -221,7 +222,7 @@ extern int mca_btl_tcp_send(
* @param endpoint (IN) BTL addressing information
* @param descriptor (IN) Description of the data to be transferred
*/
extern int mca_btl_tcp_put(
struct mca_btl_base_module_t* btl,
struct mca_btl_base_endpoint_t* btl_peer,
@ -236,7 +237,7 @@ extern int mca_btl_tcp_put(
* @param endpoint (IN) BTL addressing information
* @param descriptor (IN) Description of the data to be transferred
*/
extern int mca_btl_tcp_get(
struct mca_btl_base_module_t* btl,
struct mca_btl_base_endpoint_t* btl_peer,

16
ompi/mca/btl/tcp/btl_tcp_component.c
Просмотреть файл

@ -207,20 +207,22 @@ int mca_btl_tcp_component_open(void)
mca_btl_tcp_param_register_int ("sndbuf", 128*1024);
mca_btl_tcp_component.tcp_rcvbuf =
mca_btl_tcp_param_register_int ("rcvbuf", 128*1024);
mca_btl_tcp_component.tcp_endpoint_cache =
mca_btl_tcp_param_register_int ("endpoint_cache", 30*1024);
mca_btl_tcp_module.super.btl_exclusivity =
mca_btl_tcp_param_register_int ("exclusivity", MCA_BTL_EXCLUSIVITY_LOW);
mca_btl_tcp_module.super.btl_eager_limit =
mca_btl_tcp_module.super.btl_eager_limit =
mca_btl_tcp_param_register_int ("eager_limit", 64*1024);
mca_btl_tcp_module.super.btl_min_send_size =
mca_btl_tcp_param_register_int ("min_send_size", 64*1024);
mca_btl_tcp_module.super.btl_max_send_size =
mca_btl_tcp_param_register_int ("max_send_size", 128*1024);
mca_btl_tcp_module.super.btl_min_rdma_size =
mca_btl_tcp_param_register_int("min_rdma_size", 128*1024);
mca_btl_tcp_module.super.btl_max_rdma_size =
mca_btl_tcp_param_register_int("max_rdma_size", INT_MAX);
mca_btl_tcp_module.super.btl_flags =
mca_btl_tcp_param_register_int("flags", MCA_BTL_FLAGS_PUT|MCA_BTL_FLAGS_SEND_INPLACE);
mca_btl_tcp_module.super.btl_min_rdma_size =
mca_btl_tcp_param_register_int("min_rdma_size", 128*1024);
mca_btl_tcp_module.super.btl_max_rdma_size =
mca_btl_tcp_param_register_int("max_rdma_size", INT_MAX);
mca_btl_tcp_module.super.btl_flags =
mca_btl_tcp_param_register_int("flags", MCA_BTL_FLAGS_PUT|MCA_BTL_FLAGS_SEND_INPLACE);
return OMPI_SUCCESS;
}

103
ompi/mca/btl/tcp/btl_tcp_endpoint.c
Просмотреть файл

@ -73,6 +73,11 @@ static void mca_btl_tcp_endpoint_construct(mca_btl_tcp_endpoint_t* endpoint)
endpoint->endpoint_state = MCA_BTL_TCP_CLOSED;
endpoint->endpoint_retries = 0;
endpoint->endpoint_nbo = false;
#if MCA_BTL_TCP_ENDPOINT_CACHE
endpoint->endpoint_cache = NULL;
endpoint->endpoint_cache_pos = 0;
endpoint->endpoint_cache_length = 0;
#endif /* MCA_BTL_TCP_ENDPOINT_CACHE */
OBJ_CONSTRUCT(&endpoint->endpoint_frags, opal_list_t);
OBJ_CONSTRUCT(&endpoint->endpoint_send_lock, opal_mutex_t);
OBJ_CONSTRUCT(&endpoint->endpoint_recv_lock, opal_mutex_t);
@ -177,6 +182,10 @@ static void mca_btl_tcp_endpoint_dump(mca_btl_base_endpoint_t* btl_endpoint, con
static inline void mca_btl_tcp_endpoint_event_init(mca_btl_base_endpoint_t* btl_endpoint, int sd)
{
#if MCA_BTL_TCP_ENDPOINT_CACHE
btl_endpoint->endpoint_cache = (char*)malloc(mca_btl_tcp_component.tcp_endpoint_cache);
#endif /* MCA_BTL_TCP_ENDPOINT_CACHE */
opal_event_set(
&btl_endpoint->endpoint_recv_event,
btl_endpoint->endpoint_sd,
@ -342,6 +351,10 @@ void mca_btl_tcp_endpoint_close(mca_btl_base_endpoint_t* btl_endpoint)
opal_event_del(&btl_endpoint->endpoint_send_event);
close(btl_endpoint->endpoint_sd);
btl_endpoint->endpoint_sd = -1;
#if MCA_BTL_TCP_ENDPOINT_CACHE
free( btl_endpoint->endpoint_cache );
btl_endpoint->endpoint_cache = NULL;
#endif /* MCA_BTL_TCP_ENDPOINT_CACHE */
}
btl_endpoint->endpoint_state = MCA_BTL_TCP_CLOSED;
btl_endpoint->endpoint_retries++;
@ -501,7 +514,7 @@ static int mca_btl_tcp_endpoint_start_connect(mca_btl_base_endpoint_t* btl_endpo
if(fcntl(btl_endpoint->endpoint_sd, F_SETFL, flags) < 0)
BTL_ERROR(("fcntl(F_SETFL) failed with errno=%d", ompi_socket_errno));
}
/* start the connect - will likely fail with EINPROGRESS */
endpoint_addr.sin_family = AF_INET;
endpoint_addr.sin_addr = btl_endpoint->endpoint_addr->addr_inet;
@ -582,57 +595,67 @@ static void mca_btl_tcp_endpoint_recv_handler(int sd, short flags, void* user)
switch(btl_endpoint->endpoint_state) {
case MCA_BTL_TCP_CONNECT_ACK:
{
mca_btl_tcp_endpoint_recv_connect_ack(btl_endpoint);
OPAL_THREAD_UNLOCK(&btl_endpoint->endpoint_recv_lock);
break;
mca_btl_tcp_endpoint_recv_connect_ack(btl_endpoint);
OPAL_THREAD_UNLOCK(&btl_endpoint->endpoint_recv_lock);
break;
}
case MCA_BTL_TCP_CONNECTED:
{
mca_btl_tcp_frag_t* frag = btl_endpoint->endpoint_recv_frag;
if(NULL == frag) {
int rc;
MCA_BTL_TCP_FRAG_ALLOC_MAX(frag, rc);
if(NULL == frag) {
OPAL_THREAD_UNLOCK(&btl_endpoint->endpoint_recv_lock);
return;
}
MCA_BTL_TCP_FRAG_INIT_DST(frag, btl_endpoint);
}
mca_btl_tcp_frag_t* frag;
/* check for completion of non-blocking recv on the current fragment */
if(mca_btl_tcp_frag_recv(frag, sd) == false) {
btl_endpoint->endpoint_recv_frag = frag;
OPAL_THREAD_UNLOCK(&btl_endpoint->endpoint_recv_lock);
} else {
btl_endpoint->endpoint_recv_frag = NULL;
OPAL_THREAD_UNLOCK(&btl_endpoint->endpoint_recv_lock);
switch(frag->hdr.type) {
case MCA_BTL_TCP_HDR_TYPE_SEND:
{
mca_btl_base_recv_reg_t* reg = frag->btl->tcp_reg + frag->hdr.base.tag;
reg->cbfunc(&frag->btl->super, frag->hdr.base.tag, &frag->base, reg->cbdata);
break;
}
default:
{
break;
#if MCA_BTL_TCP_ENDPOINT_CACHE
btl_endpoint->endpoint_cache_pos = 0;
data_still_pending_on_endpoint:
#endif /* MCA_BTL_TCP_ENDPOINT_CACHE */
frag = btl_endpoint->endpoint_recv_frag;
if(NULL == frag) {
int rc;
MCA_BTL_TCP_FRAG_ALLOC_MAX(frag, rc);
if(NULL == frag) {
OPAL_THREAD_UNLOCK(&btl_endpoint->endpoint_recv_lock);
return;
}
MCA_BTL_TCP_FRAG_INIT_DST(frag, btl_endpoint);
}
MCA_BTL_TCP_FRAG_RETURN_MAX(frag);
}
break;
/* check for completion of non-blocking recv on the current fragment */
if(mca_btl_tcp_frag_recv(frag, sd) == false) {
btl_endpoint->endpoint_recv_frag = frag;
OPAL_THREAD_UNLOCK(&btl_endpoint->endpoint_recv_lock);
} else {
btl_endpoint->endpoint_recv_frag = NULL;
OPAL_THREAD_UNLOCK(&btl_endpoint->endpoint_recv_lock);
switch(frag->hdr.type) {
case MCA_BTL_TCP_HDR_TYPE_SEND:
{
mca_btl_base_recv_reg_t* reg = frag->btl->tcp_reg + frag->hdr.base.tag;
reg->cbfunc(&frag->btl->super, frag->hdr.base.tag, &frag->base, reg->cbdata);
break;
}
default:
{
break;
}
}
MCA_BTL_TCP_FRAG_RETURN_MAX(frag);
#if MCA_BTL_TCP_ENDPOINT_CACHE
if( 0 != btl_endpoint->endpoint_cache_length )
goto data_still_pending_on_endpoint;
#endif /* MCA_BTL_TCP_ENDPOINT_CACHE */
}
break;
}
case MCA_BTL_TCP_SHUTDOWN:
{
OPAL_THREAD_UNLOCK(&btl_endpoint->endpoint_recv_lock);
break;
OPAL_THREAD_UNLOCK(&btl_endpoint->endpoint_recv_lock);
break;
}
default:
{
OPAL_THREAD_UNLOCK(&btl_endpoint->endpoint_recv_lock);
BTL_ERROR(("invalid socket state(%d)", btl_endpoint->endpoint_state));
mca_btl_tcp_endpoint_close(btl_endpoint);
break;
OPAL_THREAD_UNLOCK(&btl_endpoint->endpoint_recv_lock);
BTL_ERROR(("invalid socket state(%d)", btl_endpoint->endpoint_state));
mca_btl_tcp_endpoint_close(btl_endpoint);
break;
}
}
}

34
ompi/mca/btl/tcp/btl_tcp_endpoint.h
Просмотреть файл

@ -31,6 +31,7 @@ extern "C" {
OBJ_CLASS_DECLARATION(mca_btl_tcp_endpoint_t);
#define MCA_BTL_TCP_ENDPOINT_CACHE 1
/**
* State of TCP endpoint connection.
@ -54,20 +55,25 @@ typedef enum {
struct mca_btl_base_endpoint_t {
opal_list_item_t super;
struct mca_btl_tcp_module_t* endpoint_btl; /**< BTL instance that created this connection */
struct mca_btl_tcp_proc_t* endpoint_proc; /**< proc structure corresponding to endpoint */
struct mca_btl_tcp_addr_t* endpoint_addr; /**< address of endpoint */
int endpoint_sd; /**< socket connection to endpoint */
struct mca_btl_tcp_frag_t* endpoint_send_frag; /**< current send frag being processed */
struct mca_btl_tcp_frag_t* endpoint_recv_frag; /**< current recv frag being processed */
mca_btl_tcp_state_t endpoint_state; /**< current state of the connection */
size_t endpoint_retries; /**< number of connection retries attempted */
opal_list_t endpoint_frags; /**< list of pending frags to send */
opal_mutex_t endpoint_send_lock; /**< lock for concurrent access to endpoint state */
opal_mutex_t endpoint_recv_lock; /**< lock for concurrent access to endpoint state */
opal_event_t endpoint_send_event; /**< event for async processing of send frags */
opal_event_t endpoint_recv_event; /**< event for async processing of recv frags */
bool endpoint_nbo; /**< convert headers to network byte order? */
struct mca_btl_tcp_module_t* endpoint_btl; /**< BTL instance that created this connection */
struct mca_btl_tcp_proc_t* endpoint_proc; /**< proc structure corresponding to endpoint */
struct mca_btl_tcp_addr_t* endpoint_addr; /**< address of endpoint */
int endpoint_sd; /**< socket connection to endpoint */
#if MCA_BTL_TCP_ENDPOINT_CACHE
char* endpoint_cache; /**< cache for the recv (reduce the number of recv syscall */
size_t endpoint_cache_pos; /**< */
size_t endpoint_cache_length; /**< */
#endif /* MCA_BTL_TCP_ENDPOINT_CACHE */
struct mca_btl_tcp_frag_t* endpoint_send_frag; /**< current send frag being processed */
struct mca_btl_tcp_frag_t* endpoint_recv_frag; /**< current recv frag being processed */
mca_btl_tcp_state_t endpoint_state; /**< current state of the connection */
size_t endpoint_retries; /**< number of connection retries attempted */
opal_list_t endpoint_frags; /**< list of pending frags to send */
opal_mutex_t endpoint_send_lock; /**< lock for concurrent access to endpoint state */
opal_mutex_t endpoint_recv_lock; /**< lock for concurrent access to endpoint state */
opal_event_t endpoint_send_event; /**< event for async processing of send frags */
opal_event_t endpoint_recv_event; /**< event for async processing of recv frags */
bool endpoint_nbo; /**< convert headers to network byte order? */
};
typedef struct mca_btl_base_endpoint_t mca_btl_base_endpoint_t;

112
ompi/mca/btl/tcp/btl_tcp_frag.c
Просмотреть файл

@ -94,7 +94,7 @@ bool mca_btl_tcp_frag_send(mca_btl_tcp_frag_t* frag, int sd)
}
}
}
/* if the write didn't complete - update the iovec state */
num_vecs = frag->iov_cnt;
for(i=0; i<num_vecs; i++) {
@ -117,14 +117,38 @@ bool mca_btl_tcp_frag_send(mca_btl_tcp_frag_t* frag, int sd)
bool mca_btl_tcp_frag_recv(mca_btl_tcp_frag_t* frag, int sd)
{
int cnt;
size_t i, num_vecs;
size_t i, num_vecs = frag->iov_cnt;
mca_btl_base_endpoint_t* btl_endpoint = frag->endpoint;
repeat:
repeat:
#if MCA_BTL_TCP_ENDPOINT_CACHE
if( 0 != btl_endpoint->endpoint_cache_length ) {
size_t length = btl_endpoint->endpoint_cache_length;
cnt = btl_endpoint->endpoint_cache_length;
for( i = 0; i < frag->iov_cnt; i++ ) {
if( length > frag->iov_ptr[i].iov_len )
length = frag->iov_ptr[0].iov_len;
memcpy( frag->iov_ptr[i].iov_base,
btl_endpoint->endpoint_cache + btl_endpoint->endpoint_cache_pos,
length );
btl_endpoint->endpoint_cache_pos += length;
btl_endpoint->endpoint_cache_length -= length;
length = btl_endpoint->endpoint_cache_length;
if( 0 == length ) break;
}
goto advance_iov_position;
}
frag->iov_ptr[num_vecs].iov_base = btl_endpoint->endpoint_cache;
frag->iov_ptr[num_vecs].iov_len = mca_btl_tcp_component.tcp_endpoint_cache;
num_vecs++;
#endif /* MCA_BTL_TCP_ENDPOINT_CACHE */
/* non-blocking read, but continue if interrupted */
cnt = -1;
while(cnt < 0) {
cnt = readv(sd, frag->iov_ptr, frag->iov_cnt);
while( cnt < 0 ) {
cnt = readv(sd, frag->iov_ptr, num_vecs);
if(cnt < 0) {
switch(ompi_socket_errno) {
case EINTR:
@ -133,23 +157,25 @@ repeat:
return false;
case EFAULT:
opal_output( 0, "mca_btl_tcp_frag_send: writev error (%p, %d)\n\t%s(%d)\n",
frag->iov_ptr[0].iov_base, frag->iov_ptr[0].iov_len,
strerror(ompi_socket_errno), frag->iov_cnt );
frag->iov_ptr[0].iov_base, frag->iov_ptr[0].iov_len,
strerror(ompi_socket_errno), frag->iov_cnt );
default:
{
opal_output(0, "mca_btl_tcp_frag_send: writev failed with errno=%d",
ompi_socket_errno);
mca_btl_tcp_endpoint_close(frag->endpoint);
return false;
opal_output(0, "mca_btl_tcp_frag_send: writev failed with errno=%d",
ompi_socket_errno);
mca_btl_tcp_endpoint_close(btl_endpoint);
return false;
}
}
}
if(cnt == 0) {
mca_btl_tcp_endpoint_close(frag->endpoint);
mca_btl_tcp_endpoint_close(btl_endpoint);
return false;
}
}
goto advance_iov_position;
};
advance_iov_position:
/* if the write didn't complete - update the iovec state */
num_vecs = frag->iov_cnt;
for(i=0; i<num_vecs; i++) {
@ -162,45 +188,49 @@ repeat:
frag->iov_ptr->iov_base = (ompi_iov_base_ptr_t)
(((unsigned char*)frag->iov_ptr->iov_base) + cnt);
frag->iov_ptr->iov_len -= cnt;
cnt = 0;
break;
}
}
#if MCA_BTL_TCP_ENDPOINT_CACHE
btl_endpoint->endpoint_cache_length = cnt;
#endif /* MCA_BTL_TCP_ENDPOINT_CACHE */
/* read header */
if(frag->iov_cnt == 0) {
switch(frag->hdr.type) {
case MCA_BTL_TCP_HDR_TYPE_SEND:
if(frag->iov_idx == 1 && frag->hdr.size) {
frag->iov[1].iov_base = (void*)(frag+1);
frag->iov[1].iov_len = frag->hdr.size;
frag->segments[0].seg_addr.pval = frag+1;
frag->segments[0].seg_len = frag->hdr.size;
case MCA_BTL_TCP_HDR_TYPE_SEND:
if(frag->iov_idx == 1 && frag->hdr.size) {
frag->iov[1].iov_base = (void*)(frag+1);
frag->iov[1].iov_len = frag->hdr.size;
frag->segments[0].seg_addr.pval = frag+1;
frag->segments[0].seg_len = frag->hdr.size;
frag->iov_cnt++;
goto repeat;
}
break;
case MCA_BTL_TCP_HDR_TYPE_PUT:
if(frag->iov_idx == 1) {
frag->iov[1].iov_base = (void*)frag->segments;
frag->iov[1].iov_len = frag->hdr.count * sizeof(mca_btl_base_segment_t);
frag->iov_cnt++;
goto repeat;
} else if (frag->iov_idx == 2) {
for(i=0; i<frag->hdr.count; i++) {
frag->iov[i+2].iov_base = frag->segments[i].seg_addr.pval;
frag->iov[i+2].iov_len = frag->segments[i].seg_len;
frag->iov_cnt++;
goto repeat;
}
break;
case MCA_BTL_TCP_HDR_TYPE_PUT:
if(frag->iov_idx == 1) {
frag->iov[1].iov_base = (void*)frag->segments;
frag->iov[1].iov_len = frag->hdr.count * sizeof(mca_btl_base_segment_t);
frag->iov_cnt++;
goto repeat;
} else if (frag->iov_idx == 2) {
for(i=0; i<frag->hdr.count; i++) {
frag->iov[i+2].iov_base = frag->segments[i].seg_addr.pval;
frag->iov[i+2].iov_len = frag->segments[i].seg_len;
frag->iov_cnt++;
}
goto repeat;
}
break;
case MCA_BTL_TCP_HDR_TYPE_GET:
default:
break;
goto repeat;
}
break;
case MCA_BTL_TCP_HDR_TYPE_GET:
default:
break;
}
return true;
} else {
return false;
}
return false;
}