66 |
socklen_t mpi_tmp_namelen = 0; |
socklen_t mpi_tmp_namelen = 0; |
67 |
struct sockaddr mpi_tmp_name; |
struct sockaddr mpi_tmp_name; |
68 |
int retval = getpeername( socket, &mpi_tmp_name, &mpi_tmp_namelen ); |
int retval = getpeername( socket, &mpi_tmp_name, &mpi_tmp_namelen ); |
69 |
if (retval == -1) {printf("errno: %d: \n", errno); perror("getpeername");} |
if (retval == -1) {fprintf(stderr, "errno: %d: \n", errno); perror("getpeername");} |
70 |
return 1 + getpeername( socket, &mpi_tmp_name, &mpi_tmp_namelen ); |
return 1 + getpeername( socket, &mpi_tmp_name, &mpi_tmp_namelen ); |
71 |
} |
} |
72 |
|
|
76 |
// to set up the initial connection, or to create a new one if the |
// to set up the initial connection, or to create a new one if the |
77 |
// old one dies. |
// old one dies. |
78 |
|
|
79 |
static void list_accept_new_socket(list_sd) |
static void list_accept_new_socket(int list_sd) |
|
int list_sd; |
|
80 |
{ struct sockaddr_in new_sin; |
{ struct sockaddr_in new_sin; |
81 |
int ns, source; |
int ns, source; |
82 |
socklen_t fromlen = sizeof(new_sin); |
socklen_t fromlen = sizeof(new_sin); |
106 |
} |
} |
107 |
} |
} |
108 |
|
|
109 |
static int list_connect_new_socket( myrank, dest_rank ) |
static int list_connect_new_socket( int myrank, int dest_rank ) |
|
int myrank, dest_rank; |
|
110 |
{ int sd; |
{ int sd; |
111 |
char buf[sizeof(long)]; |
char buf[sizeof(long)]; |
112 |
|
|
184 |
/* Under POSIX, the system calls used here that are interruptible (EINTR) |
/* Under POSIX, the system calls used here that are interruptible (EINTR) |
185 |
are: connect(), recv(), select() |
are: connect(), recv(), select() |
186 |
*/ |
*/ |
187 |
static int select_block_sigs( n, readfds, writefds, exceptfds, timeout ) |
static int select_block_sigs( int n, fd_set *readfds, fd_set *writefds, |
188 |
int n; |
fd_set *exceptfds, struct timeval *timeout ) |
|
fd_set *readfds; |
|
|
fd_set *writefds; |
|
|
fd_set *exceptfds; |
|
|
struct timeval *timeout; |
|
189 |
{ sigset_t sigset, oldsigset; |
{ sigset_t sigset, oldsigset; |
190 |
sighandler_t real_handler; |
sighandler_t real_handler; |
191 |
int answer; |
int answer; |
395 |
#define DONT_EAT_HDR 0 |
#define DONT_EAT_HDR 0 |
396 |
#define BLOCKING 0 |
#define BLOCKING 0 |
397 |
#define NOT_BLOCKING 1 |
#define NOT_BLOCKING 1 |
398 |
|
// These next two are the values of flag in MPI_Iprobe, and must be 1 and 0 |
399 |
#define MSG_WAITING 1 |
#define MSG_WAITING 1 |
400 |
#define NO_MSG_WAITING 0 |
#define NO_MSG_WAITING 0 |
401 |
MPI_Status dummy_status; |
MPI_Status dummy_status; |
402 |
|
|
403 |
static struct timeval * read_msg_hdr_setup( int *source, int tag, int *flag ) { |
static struct timeval * read_msg_hdr_setup( int source, int tag, |
404 |
|
int blocking ) { |
405 |
static struct timeval zerotime; |
static struct timeval zerotime; |
406 |
/* MUST ALSO HANDLE MPI_ANY_TAG and MPI_COLL_COMM_TAG */ |
/* MUST ALSO HANDLE MPI_ANY_TAG and MPI_COLL_COMM_TAG */ |
407 |
if ( *source > MPINU_num_slaves || *source < MPI_ANY_SOURCE || |
if ( source > MPINU_num_slaves || source < MPI_ANY_SOURCE || |
408 |
tag < MPI_COLL_COMM_TAG ) { |
tag < MPI_COLL_COMM_TAG ) { |
409 |
printf("read_msg_hdr: invalid argument\n"); |
printf("read_msg_hdr: invalid argument\n"); |
410 |
exit(1); |
exit(1); |
411 |
} |
} |
412 |
if ( *flag == BLOCKING ) |
if ( blocking == BLOCKING ) |
413 |
return NULL; /* NULL timeout means in select means wait forever (block) */ |
return NULL; /* NULL timeout means in select means wait forever (block) */ |
414 |
else { /* poll and return */ |
else { /* poll and return */ |
415 |
zerotime.tv_sec = 0; |
zerotime.tv_sec = 0; |
418 |
} |
} |
419 |
} |
} |
420 |
|
|
421 |
static int read_msg_hdr_select(int *source, fd_set *fd_readset_ptr, int *flag, |
static int read_msg_hdr_select(int source, int tag, fd_set *fd_readset_ptr, |
422 |
struct timeval *timeout) { |
int *blocking, struct timeval *timeout, int *rank) { |
423 |
int tmp, fd, num_sd; |
int tmp, fd, num_sd; |
424 |
tmp = MPINU_rank_of_msg_avail_in_cache(*source); |
tmp = MPINU_rank_of_msg_avail_in_cache(source, tag); |
425 |
if (tmp != -1) { |
if (tmp != -1) { |
426 |
*source = tmp; /* in case *source was MPI_ANY_SOURCE */ |
*rank = tmp; /* in case source was MPI_ANY_SOURCE */ |
427 |
*flag = MSG_WAITING; |
*blocking = MSG_WAITING; |
428 |
return MPINU_pg_array[tmp].sd; |
return MPINU_pg_array[tmp].sd; |
429 |
} |
} else |
430 |
if (*source == MPINU_myrank) { /* If want msg from self and not in cache ... */ |
*rank = source; |
431 |
if ( *flag == BLOCKING ) { |
if (source == MPINU_myrank) { /* If want msg from self and not in cache ... */ |
432 |
|
if ( *blocking == BLOCKING ) { |
433 |
printf("read_msg_hdr: rank %d BLOCKING waiting for msg from self.\n", |
printf("read_msg_hdr: rank %d BLOCKING waiting for msg from self.\n", |
434 |
MPINU_myrank); |
MPINU_myrank); |
435 |
exit(1); |
exit(1); |
436 |
} else { |
} else { |
437 |
*flag = NO_MSG_WAITING; /* no messages, return */ |
*blocking = NO_MSG_WAITING; /* no messages, return */ |
438 |
return -1; |
return -1; |
439 |
} |
} |
440 |
} |
} |
441 |
|
|
442 |
if ( *source == MPI_ANY_SOURCE ) { |
if ( source == MPI_ANY_SOURCE ) { |
443 |
/* Copy global MPINU_fdset (with everything set) to local fd_readset */ |
/* Copy global MPINU_fdset (with everything set) to local fd_readset */ |
444 |
memcpy( (char *)fd_readset_ptr, (char *)&MPINU_fdset, |
memcpy( (char *)fd_readset_ptr, (char *)&MPINU_fdset, |
445 |
sizeof(MPINU_fdset) ); |
sizeof(MPINU_fdset) ); |
446 |
} else { |
} else { |
447 |
fd = MPINU_pg_array[*source].sd; |
fd = MPINU_pg_array[source].sd; |
448 |
FD_ZERO( fd_readset_ptr ); |
FD_ZERO( fd_readset_ptr ); |
449 |
if ( fd != PG_NOSOCKET ) |
if ( fd != PG_NOSOCKET ) |
450 |
FD_SET( fd, fd_readset_ptr ); |
FD_SET( fd, fd_readset_ptr ); |
464 |
CALL_CHK( num_sd = select_block_sigs, (MPINU_max_sd+1, fd_readset_ptr, |
CALL_CHK( num_sd = select_block_sigs, (MPINU_max_sd+1, fd_readset_ptr, |
465 |
NULL, NULL, timeout) ); |
NULL, NULL, timeout) ); |
466 |
if ( FD_ISSET( MPINU_my_list_sd, fd_readset_ptr ) ) { |
if ( FD_ISSET( MPINU_my_list_sd, fd_readset_ptr ) ) { |
467 |
assert( *source != MPI_ANY_SOURCE ); |
assert( source != MPI_ANY_SOURCE ); |
468 |
/* A msg on this socket is a request to open a new socket. */ |
/* A msg on this socket is a request to open a new socket. */ |
469 |
list_accept_new_socket(MPINU_my_list_sd); |
list_accept_new_socket(MPINU_my_list_sd); |
470 |
// If after list_accept_..., there's now a socket for *source, register it; |
// If after list_accept_..., there's now a socket for source, register it; |
471 |
// Else it was a new socket, but not one for *source, and do nothing |
// Else it was a new socket, but not one for source, and do nothing |
472 |
if (MPINU_pg_array[*source].sd != PG_NOSOCKET) |
if (MPINU_pg_array[source].sd != PG_NOSOCKET) |
473 |
FD_SET( MPINU_pg_array[*source].sd, fd_readset_ptr ); |
FD_SET( MPINU_pg_array[source].sd, fd_readset_ptr ); |
474 |
errno = EAGAIN; |
errno = EAGAIN; |
475 |
return -1; |
return -1; |
476 |
} |
} |
477 |
|
|
478 |
/* Given *flag with value of BLOCKING or NOT_BLOCKING, |
/* Given *blocking with value of BLOCKING or NOT_BLOCKING, |
479 |
change to MSG_WAITING or NO_MSG_WAITING */ |
change to MSG_WAITING or NO_MSG_WAITING */ |
480 |
if ( num_sd == 0 ) { |
if ( num_sd == 0 ) { |
481 |
if ( *flag == BLOCKING ) { |
if ( *blocking == BLOCKING ) { |
482 |
printf("read_msg_hdr: rank %d BLOCKING exiting without recv'd msg.\n", |
printf("read_msg_hdr: rank %d BLOCKING exiting without recv'd msg.\n", |
483 |
MPINU_myrank); |
MPINU_myrank); |
484 |
exit(1); |
exit(1); |
485 |
} else { |
} else { |
486 |
*flag = NO_MSG_WAITING; /* no messages, return */ |
*blocking = NO_MSG_WAITING; /* no messages, return */ |
487 |
return -1; |
return -1; |
488 |
} |
} |
489 |
} else if ( num_sd > 0 ) |
} else if ( num_sd > 0 ) |
490 |
*flag = MSG_WAITING; |
*blocking = MSG_WAITING; |
491 |
|
|
492 |
return num_sd; |
return num_sd; |
493 |
} |
} |
494 |
|
|
495 |
static struct msg_hdr * read_msg_hdr_get_buf(int *source, int tag, |
static struct msg_hdr * read_msg_hdr_get_buf(int source, int tag, |
496 |
fd_set *fd_readset_ptr, int eat_hdr) { |
fd_set *fd_readset_ptr, int eat_hdr) { |
497 |
int fd, recv_flags, num_bytes, tmp; |
int fd, recv_flags, num_bytes, dest; |
498 |
static struct msg_hdr buf_hdr; // static since we return this to caller |
static struct msg_hdr buf_hdr; // static since we return this to caller |
499 |
|
|
500 |
if ( eat_hdr == DONT_EAT_HDR ) recv_flags = MSG_PEEK; |
if ( eat_hdr == DONT_EAT_HDR ) recv_flags = MSG_PEEK; |
501 |
else recv_flags = 0; |
else recv_flags = 0; |
502 |
|
|
503 |
if ( *source == MPI_ANY_SOURCE ) { /* Determine source, fd */ |
if ( source == MPI_ANY_SOURCE ) { /* Determine source, fd */ |
504 |
FD_CLR( MPINU_my_list_sd, fd_readset_ptr ); /* listener not legit source */ |
FD_CLR( MPINU_my_list_sd, fd_readset_ptr ); /* listener not legit source */ |
505 |
for ( fd = 0; fd <= MPINU_max_sd+1; fd++) |
for ( fd = 0; fd <= MPINU_max_sd+1; fd++) |
506 |
if ( FD_ISSET( fd, fd_readset_ptr ) ) break; |
if ( FD_ISSET( fd, fd_readset_ptr ) ) break; |
509 |
exit(1); |
exit(1); |
510 |
} |
} |
511 |
} else |
} else |
512 |
fd = MPINU_pg_array[*source].sd; |
fd = MPINU_pg_array[source].sd; |
513 |
#ifdef DEBUG |
#ifdef DEBUG |
514 |
printf("receiving on socket %d out of MPINU_max_sd %d\n", fd, MPINU_max_sd); |
printf("receiving on socket %d out of MPINU_max_sd %d\n", fd, MPINU_max_sd); |
515 |
#endif |
#endif |
530 |
return NULL; /* failed to receive; return NULL and try again upstairs */ |
return NULL; /* failed to receive; return NULL and try again upstairs */ |
531 |
} |
} |
532 |
// Confirm that the source for that socket is the same as source in buf_hdr |
// Confirm that the source for that socket is the same as source in buf_hdr |
533 |
for ( tmp = 0; tmp <= MPINU_num_slaves; tmp++ ) |
for ( dest = 0; dest <= MPINU_num_slaves; dest++ ) |
534 |
if ( MPINU_pg_array[tmp].sd == fd ) |
if ( MPINU_pg_array[dest].sd == fd ) |
535 |
assert( tmp == ntohl(buf_hdr.rank) ); |
assert( dest == ntohl(buf_hdr.rank) ); |
536 |
return &buf_hdr; |
return &buf_hdr; |
537 |
} |
} |
538 |
|
|
539 |
void read_msg_hdr_post_check( int eat_hdr, int *source, int tag, int size, |
static void read_msg_hdr_post_check( int eat_hdr, int source, int tag, |
540 |
MPI_Datatype datatype, int count, |
int size, MPI_Datatype datatype, |
541 |
MPI_Status *status ) { |
int count, MPI_Status *status ) { |
542 |
assert( size != -1 ); |
assert( size != -1 ); |
543 |
if ( (tag != MPI_ANY_TAG) && (tag != status->MPI_TAG) ) { |
if ( (tag != MPI_ANY_TAG) && (tag != status->MPI_TAG) ) { |
544 |
printf("MPINU(rank %d): Msg received with tag (%d) from source (%d)\n" |
printf("MPINU(rank %d): Msg received with tag (%d) from source (%d)\n" |
560 |
exit(1); |
exit(1); |
561 |
} |
} |
562 |
} |
} |
563 |
if ( ( *source != status->MPI_SOURCE ) && ( *source != MPI_ANY_SOURCE ) ) { |
if ( ( source != status->MPI_SOURCE ) && ( source != MPI_ANY_SOURCE ) ) { |
564 |
printf("read_msg_hdr: inconsistent source,\n"); |
printf("read_msg_hdr: inconsistent source,\n"); |
565 |
printf(" expecting source(%d) and received from source(%d) on proc. %d\n", |
printf(" expecting source(%d) and received from source(%d) on proc. %d\n", |
566 |
*source, status->MPI_SOURCE, MPINU_myrank); |
source, status->MPI_SOURCE, MPINU_myrank); |
567 |
exit(1); |
exit(1); |
568 |
} |
} |
569 |
} |
} |
577 |
* eat_hdr: eat_hdr == DONT_EAT_HDR means don't consume msg header |
* eat_hdr: eat_hdr == DONT_EAT_HDR means don't consume msg header |
578 |
*/ |
*/ |
579 |
static int read_msg_hdr ( int count, MPI_Datatype datatype, |
static int read_msg_hdr ( int count, MPI_Datatype datatype, |
580 |
int *source, int tag, MPI_Comm comm, int *flag, |
int *source, int tag, MPI_Comm comm, int *blocking, |
581 |
MPI_Status *status, int eat_hdr ) { |
MPI_Status *status, int eat_hdr ) { |
582 |
int size, fd, num_bytes, num_sd; |
int size, fd, num_bytes, num_sd, rank; |
583 |
fd_set fd_readset; |
fd_set fd_readset; |
584 |
struct timeval *timeout; |
struct timeval *timeout; |
585 |
struct msg_hdr * buf_hdr = NULL; |
struct msg_hdr * buf_hdr = NULL; |
593 |
exit(1); |
exit(1); |
594 |
#endif |
#endif |
595 |
|
|
596 |
timeout = read_msg_hdr_setup( source, tag, flag ); |
timeout = read_msg_hdr_setup( *source, tag, *blocking ); |
597 |
|
|
598 |
while ( buf_hdr == NULL ) { |
while ( buf_hdr == NULL ) { |
599 |
/* Use select to find active port; if BLOCKING && not MPI_ANY_SOURCE, |
/* Use select to find active port; if BLOCKING && not MPI_ANY_SOURCE, |
601 |
that we also need to check the listener port */ |
that we also need to check the listener port */ |
602 |
/* num_sd is number of socket descriptors with available messages; |
/* num_sd is number of socket descriptors with available messages; |
603 |
* num_sd is used only for debugging */ |
* num_sd is used only for debugging */ |
604 |
while ( -1 == (num_sd = |
do { |
605 |
read_msg_hdr_select(source, &fd_readset, flag, timeout)) |
num_sd = read_msg_hdr_select(*source, tag, |
606 |
&& errno == EAGAIN ); |
&fd_readset, blocking, timeout, &rank); |
607 |
|
} while ( ( -1 == num_sd ) && ( errno == EAGAIN ) ); |
608 |
#ifdef DEBUG |
#ifdef DEBUG |
609 |
printf("exiting select(%d)\n",MPINU_myrank);fflush(stdout); |
printf("exiting select(%d)\n",MPINU_myrank);fflush(stdout); |
610 |
printf("timeout == NULL: %d; num_sd: %d; *flag: %d\n", |
printf("timeout == NULL: %d; num_sd: %d; *blocking: %d\n", |
611 |
timeout == NULL, num_sd, *flag); |
timeout == NULL, num_sd, *blocking); |
612 |
#endif |
#endif |
613 |
|
|
614 |
if (*flag == NO_MSG_WAITING) |
if (*blocking == NO_MSG_WAITING) |
615 |
return 0; /* This was MPI_Iprobe; It doesn't expect a buffer or status*/ |
return 0; /* This was MPI_Iprobe; It doesn't expect a buffer or status*/ |
616 |
buf_hdr = read_msg_hdr_get_buf(source, tag, &fd_readset, eat_hdr); |
buf_hdr = read_msg_hdr_get_buf(rank, tag, &fd_readset, eat_hdr); |
617 |
} |
} |
618 |
status->MPI_TAG = ntohl(buf_hdr->tag); |
status->MPI_TAG = ntohl(buf_hdr->tag); |
619 |
size = ntohl(buf_hdr->size); |
size = ntohl(buf_hdr->size); |
620 |
status->mpi_size = size; |
status->mpi_size = size; |
621 |
status->MPI_SOURCE = ntohl(buf_hdr->rank); |
status->MPI_SOURCE = ntohl(buf_hdr->rank); |
622 |
|
|
623 |
read_msg_hdr_post_check( eat_hdr, source, tag, size, datatype, count, |
read_msg_hdr_post_check( eat_hdr, *source, tag, size, datatype, count, |
624 |
status ); |
status ); |
625 |
|
|
626 |
#ifdef DEBUG |
#ifdef DEBUG |