/[gcl]/gcl/pargcl/src/mpinu/sendrecv.c
ViewVC logotype

Diff of /gcl/pargcl/src/mpinu/sendrecv.c

Parent Directory Parent Directory | Revision Log Revision Log | View Patch Patch

revision 1.1 by gene, Fri Oct 28 23:00:29 2005 UTC revision 1.2 by gene, Fri Dec 2 07:54:27 2005 UTC
# Line 66  static int IS_SOCKET_CONNECTED(int socke Line 66  static int IS_SOCKET_CONNECTED(int socke
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    
# Line 76  if (retval == -1) {printf("errno: %d: \n Line 76  if (retval == -1) {printf("errno: %d: \n
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);
# Line 107  int list_sd; Line 106  int list_sd;
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    
# Line 186  static void *curr_signal() Line 184  static void *curr_signal()
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;
# Line 401  if ( count * datatype_size[(int)datatype Line 395  if ( count * datatype_size[(int)datatype
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;
# Line 422  static struct timeval * read_msg_hdr_set Line 418  static struct timeval * read_msg_hdr_set
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 );
# Line 467  static int read_msg_hdr_select(int *sour Line 464  static int read_msg_hdr_select(int *sour
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;
# Line 512  static struct msg_hdr * read_msg_hdr_get Line 509  static struct msg_hdr * read_msg_hdr_get
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
# Line 533  static struct msg_hdr * read_msg_hdr_get Line 530  static struct msg_hdr * read_msg_hdr_get
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"
# Line 563  void read_msg_hdr_post_check( int eat_hd Line 560  void read_msg_hdr_post_check( int eat_hd
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  }  }
# Line 580  void read_msg_hdr_post_check( int eat_hd Line 577  void read_msg_hdr_post_check( int eat_hd
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;
# Line 596  static int read_msg_hdr ( int count, MPI Line 593  static int read_msg_hdr ( int count, MPI
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,
# Line 604  static int read_msg_hdr ( int count, MPI Line 601  static int read_msg_hdr ( int count, MPI
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

Legend:
Removed from v.1.1  
changed lines
  Added in v.1.2

savannah-hackers-public@gnu.org
ViewVC Help
Powered by ViewVC 1.1.26