socki_util.i

fortran并行计算包

#   endif

    /*
     * Initialize newly allocated sock structure and associated poll structures
     */
    sock_set->pollinfos[avail_elem].sock_id = (sock_set->id << 24) | avail_elem;
    sock_set->pollinfos[avail_elem].sock = sock;
    sock->sock_set = sock_set;
    sock->elem = avail_elem;

    sock_set->pollfds[avail_elem].fd = -1;
    sock_set->pollfds[avail_elem].events = 0;
    sock_set->pollfds[avail_elem].revents = 0;

#   ifdef MPICH_IS_THREADED
    {
    MPIU_THREAD_CHECK_BEGIN
	if (sock_set->pollfds_active != NULL)
	{
	    sock_set->pollfds_updated = TRUE;
	}
    MPIU_THREAD_CHECK_END
    }
#   endif
    
    *sockp = sock;
    
  fn_exit:
    MPIDI_FUNC_EXIT(MPID_STATE_MPIDU_SOCKI_SOCK_ALLOC);
    return mpi_errno;

    /* --BEGIN ERROR HANDLING-- */
  fn_fail:
    if (pollinfos != NULL)
    {
	MPIU_Free(pollinfos);
    }

    if (pollfds != NULL)
    {
	MPIU_Free(pollfds);
    }
	
    if (sock != NULL)
    {
	MPIU_Free(sock);
    }
    
    goto fn_exit;
    /* --END ERROR HANDLING-- */
}
/* end MPIDU_Socki_sock_alloc() */


#undef FUNCNAME
#define FUNCNAME MPIDU_Socki_sock_free
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
static void MPIDU_Socki_sock_free(struct MPIDU_Sock * sock)
{
    struct pollfd * pollfd = MPIDU_Socki_sock_get_pollfd(sock);
    struct pollinfo * pollinfo = MPIDU_Socki_sock_get_pollinfo(sock);
    struct MPIDU_Sock_set * sock_set = sock->sock_set;
    MPIDI_STATE_DECL(MPID_STATE_MPIDU_SOCKI_SOCK_FREE);

    MPIDI_FUNC_ENTER(MPID_STATE_MPIDU_SOCKI_SOCK_FREE);

    /* FIXME: We need an abstraction for the thread sync operations */
#   ifdef MPICH_IS_THREADED
    {
	/*
	 * Freeing a sock while Sock_wait() is blocked in poll() is not supported
	 */
	MPIU_Assert(sock_set->pollfds_active == NULL);
    }
#   endif

    /*
     * Compress poll array
     */
     /* FIXME: move last element into current position and update sock associated with last element.
     */
    if (sock->elem + 1 == sock_set->poll_array_elems)
    { 
	sock_set->poll_array_elems -= 1;
	if (sock_set->starting_elem >= sock_set->poll_array_elems)
	{
	    sock_set->starting_elem = 0;
	}
    }

    /*
     * Remove entry from the poll list and mark the entry as free
     */
    pollinfo->fd      = -1;
    pollinfo->sock    = NULL;
    pollinfo->sock_id = -1;
    pollinfo->type    = MPIDU_SOCKI_TYPE_FIRST;
    pollinfo->state   = MPIDU_SOCKI_STATE_FIRST;
#   ifdef MPICH_IS_THREADED
    {
	pollinfo->pollfd_events = 0;
    }
#   endif
		
    pollfd->fd = -1;
    pollfd->events = 0;
    pollfd->revents = 0;

    /*
     * Mark the sock as invalid so that any future use might be caught
     */
    sock->sock_set = NULL;
    sock->elem = -1;
    
    MPIU_Free(sock);
    
    MPIDI_FUNC_EXIT(MPID_STATE_MPIDU_SOCKI_SOCK_FREE);
}
/* end MPIDU_Socki_sock_free() */


#undef FUNCNAME
#define FUNCNAME MPIDU_Socki_event_enqueue
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
static int MPIDU_Socki_event_enqueue(struct pollinfo * pollinfo, MPIDU_Sock_op_t op, MPIU_Size_t num_bytes,
				     void * user_ptr, int error)
{
    struct MPIDU_Sock_set * sock_set = pollinfo->sock_set;
    struct MPIDU_Socki_eventq_elem * eventq_elem;
    int mpi_errno = MPI_SUCCESS;
    MPIDI_STATE_DECL(MPID_STATE_SOCKI_EVENT_ENQUEUE);

    MPIDI_FUNC_ENTER(MPID_STATE_SOCKI_EVENT_ENQUEUE);

    if (MPIDU_Socki_eventq_pool != NULL)
    {
	eventq_elem = MPIDU_Socki_eventq_pool;
	MPIDU_Socki_eventq_pool = MPIDU_Socki_eventq_pool->next;
    }
    else
    {
	int i;
	struct MPIDU_Socki_eventq_table *eventq_table;

	eventq_table = MPIU_Malloc(sizeof(struct MPIDU_Socki_eventq_table));
	/* --BEGIN ERROR HANDLING-- */
	if (eventq_table == NULL)
	{
	    mpi_errno = MPIR_Err_create_code(errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER,
					     "**sock|poll|eqmalloc", 0);
	    goto fn_exit;
	}
	/* --END ERROR HANDLING-- */

        eventq_elem = eventq_table->elems;

        eventq_table->next = MPIDU_Socki_eventq_table_head;
        MPIDU_Socki_eventq_table_head = eventq_table;

	if (MPIDU_SOCK_EVENTQ_POOL_SIZE > 1)
	{ 
	    MPIDU_Socki_eventq_pool = &eventq_elem[1];
	    for (i = 0; i < MPIDU_SOCK_EVENTQ_POOL_SIZE - 2; i++)
	    {
		MPIDU_Socki_eventq_pool[i].next = &MPIDU_Socki_eventq_pool[i+1];
	    }
	    MPIDU_Socki_eventq_pool[MPIDU_SOCK_EVENTQ_POOL_SIZE - 2].next = NULL;
	}
    }
    
    eventq_elem->event.op_type = op;
    eventq_elem->event.num_bytes = num_bytes;
    eventq_elem->event.user_ptr = user_ptr;
    eventq_elem->event.error = error;
    eventq_elem->set_elem = pollinfo->elem;
    eventq_elem->next = NULL;

    if (sock_set->eventq_head == NULL)
    { 
	sock_set->eventq_head = eventq_elem;
    }
    else
    {
	sock_set->eventq_tail->next = eventq_elem;
    }
    sock_set->eventq_tail = eventq_elem;
fn_exit:
    MPIDI_FUNC_EXIT(MPID_STATE_SOCKI_EVENT_ENQUEUE);
    return mpi_errno;
}
/* end MPIDU_Socki_event_enqueue() */


#undef FUNCNAME
#define FUNCNAME MPIDU_Socki_event_dequeue
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
static inline int MPIDU_Socki_event_dequeue(struct MPIDU_Sock_set * sock_set, int * set_elem, struct MPIDU_Sock_event * eventp)
{
    struct MPIDU_Socki_eventq_elem * eventq_elem;
    int mpi_errno = MPI_SUCCESS;
    MPIDI_STATE_DECL(MPID_STATE_SOCKI_EVENT_DEQUEUE);

    MPIDI_FUNC_ENTER(MPID_STATE_SOCKI_EVENT_DEQUEUE);

    if (sock_set->eventq_head != NULL)
    {
	eventq_elem = sock_set->eventq_head;
	
	sock_set->eventq_head = eventq_elem->next;
	if (eventq_elem->next == NULL)
	{
	    sock_set->eventq_tail = NULL;
	}
	
	*eventp = eventq_elem->event;
	*set_elem = eventq_elem->set_elem;
	
	eventq_elem->next = MPIDU_Socki_eventq_pool;
	MPIDU_Socki_eventq_pool = eventq_elem;
    }
    /* --BEGIN ERROR HANDLING-- */
    else
    {
	/* FIXME: Shouldn't this be an mpi error code? */
	mpi_errno = MPIDU_SOCK_ERR_FAIL;
    }
    /* --END ERROR HANDLING-- */

    MPIDI_FUNC_EXIT(MPID_STATE_SOCKI_EVENT_DEQUEUE);
    return mpi_errno;
}
/* end MPIDU_Socki_event_dequeue() */


/* FIXME: Who allocates eventq tables?  Should there be a check that these
   tables are empty first? */
#undef FUNCNAME
#define FUNCNAME MPIDU_Socki_free_eventq_mem
#undef FCNAME
#define FCNAME "MPIDU_Socki_free_eventq_mem"
static void MPIDU_Socki_free_eventq_mem(void)
{
    struct MPIDU_Socki_eventq_table *eventq_table, *eventq_table_next;
    MPIDI_STATE_DECL(MPID_STATE_SOCKI_FREE_EVENTQ_MEM);

    MPIDI_FUNC_ENTER(MPID_STATE_SOCKI_FREE_EVENTQ_MEM);

    eventq_table = MPIDU_Socki_eventq_table_head;
    while (eventq_table) {
        eventq_table_next = eventq_table->next;
        MPIU_Free(eventq_table);
        eventq_table = eventq_table_next;
    }
    MPIDU_Socki_eventq_table_head = NULL;

    MPIDI_FUNC_EXIT(MPID_STATE_SOCKI_FREE_EVENTQ_MEM);
}

/* Provide a standard mechanism for setting the socket buffer size.
   The value is -1 if the default size hasn't been set, 0 if no size 
   should be set, and > 0 if that size should be used */
static int sockBufSize = -1;

/* Set the socket buffer sizes on fd to the standard values (this is controlled
   by the parameter MPICH_SOCK_BUFSIZE).  If "firm" is true, require that the
   sockets actually accept that buffer size.  */
int MPIDU_Sock_SetSockBufferSize( int fd, int firm )
{
    int mpi_errno = MPI_SUCCESS;
    int rc;

    /* Get the socket buffer size if we haven't yet acquired it */
    if (sockBufSize < 0) {
	/* FIXME: Is this the name that we want to use (this was chosen
	   to match the original, undocumented name) */
	rc = MPIU_GetEnvInt( "MPICH_SOCKET_BUFFER_SIZE", &sockBufSize );
	if (rc <= 0) {
	    sockBufSize = 0;
	}
	MPIU_DBG_MSG_D(CH3_CONNECT,TYPICAL,"Sock buf size = %d\n",sockBufSize);
    }

    if (sockBufSize > 0) {
	int bufsz;
	socklen_t bufsz_len;

	bufsz     = sockBufSize;
	bufsz_len = sizeof(bufsz);
	rc = setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &bufsz, bufsz_len);
	if (rc == -1) {
	    MPIU_ERR_SETANDJUMP3(mpi_errno,MPIDU_SOCK_ERR_FAIL, 
				 "**sock|poll|setsndbufsz",
				 "**sock|poll|setsndbufsz %d %d %s", 
				 bufsz, errno, MPIU_Strerror(errno));
	}
	bufsz     = sockBufSize;
	bufsz_len = sizeof(bufsz);
	rc = setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &bufsz, bufsz_len);
	if (rc == -1) {
	    MPIU_ERR_SETANDJUMP3(mpi_errno,MPIDU_SOCK_ERR_FAIL, 
				 "**sock|poll|setrcvbufsz",
				 "**sock|poll|setrcvbufsz %d %d %s", 
				 bufsz, errno, MPIU_Strerror(errno));
	}
	bufsz_len = sizeof(bufsz);

	if (firm) {
	    rc = getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &bufsz, &bufsz_len);
	    /* --BEGIN ERROR HANDLING-- */
	    if (rc == 0) {
		if (bufsz < sockBufSize * 0.9) {
		MPIU_Msg_printf("WARNING: send socket buffer size differs from requested size (requested=%d, actual=%d)\n",
				sockBufSize, bufsz);
		}
	    }
	    /* --END ERROR HANDLING-- */
	    
	    bufsz_len = sizeof(bufsz);
	    rc = getsockopt(fd, SOL_SOCKET, SO_RCVBUF, &bufsz, &bufsz_len);
	    /* --BEGIN ERROR HANDLING-- */
	    if (rc == 0) {
		if (bufsz < sockBufSize * 0.9) {
		    MPIU_Msg_printf("WARNING: receive socket buffer size differs from requested size (requested=%d, actual=%d)\n",
				    sockBufSize, bufsz);
		}
	    }
	    /* --END ERROR HANDLING-- */
	}
    }
 fn_fail:
    return mpi_errno;
}

/* This routine provides a string version of the address. */
int MPIDU_Sock_AddrToStr( MPIDU_Sock_ifaddr_t *ifaddr, char *str, int maxlen )
{ 
    int i;
    unsigned char *p = ifaddr->ifaddr;
    for (i=0; i<ifaddr->len && maxlen > 4; i++) { 
	snprintf( str, maxlen, "%.3d.", *p++ );
	str += 4;
	maxlen -= 4;
    }
    /* Change the last period to a null; but be careful in case len was zero */
    if (i > 0) *--str = 0;
    else       *str = 0;
    return 0;
}