socki_util.i

mpi并行计算的c++代码 可用vc或gcc编译通过 可以用来搭建并行计算试验环境

/* -*- Mode: C; c-basic-offset:4 ; -*- */

/*
 *  (C) 2001 by Argonne National Laboratory.
 *      See COPYRIGHT in top-level directory.
 */


#if (MPICH_THREAD_LEVEL == MPI_THREAD_MULTIPLE)
static int MPIDU_Socki_wakeup(struct MPIDU_Sock_set * sock_set);
#endif

static int MPIDU_Socki_os_to_mpi_errno(struct pollinfo * pollinfo, int os_errno, char * fcname, int line, int * conn_failed);

static int MPIDU_Socki_adjust_iov(ssize_t nb, MPID_IOV * const iov, const int count, int * const offsetp);

static int MPIDU_Socki_sock_alloc(struct MPIDU_Sock_set * sock_set, struct MPIDU_Sock ** sockp);
static void MPIDU_Socki_sock_free(struct MPIDU_Sock * sock);

static int MPIDU_Socki_event_enqueue(struct pollinfo * pollinfo, enum MPIDU_Sock_op op, MPIU_Size_t num_bytes,
				     void * user_ptr, int error);
static int inline MPIDU_Socki_event_dequeue(struct MPIDU_Sock_set * sock_set, int * set_elem, struct MPIDU_Sock_event * eventp);

static void MPIDU_Socki_free_eventq_mem(void);

struct MPIDU_Socki_eventq_table
{
    struct MPIDU_Socki_eventq_elem elems[MPIDU_SOCK_EVENTQ_POOL_SIZE];
    struct MPIDU_Socki_eventq_table * next;
};

static struct MPIDU_Socki_eventq_table *MPIDU_Socki_eventq_table_head=NULL;



#define MPIDU_Socki_sock_get_pollfd(sock_)          (&(sock_)->sock_set->pollfds[(sock_)->elem])
#define MPIDU_Socki_sock_get_pollinfo(sock_)        (&(sock_)->sock_set->pollinfos[(sock_)->elem])
#define MPIDU_Socki_pollinfo_get_pollfd(pollinfo_) (&(pollinfo_)->sock_set->pollfds[(pollinfo_)->elem])


/* Enqueue a new event.  If the enqueue fails, generate an error and jump to the fail_label_ */
#define MPIDU_SOCKI_EVENT_ENQUEUE(pollinfo_, op_, nb_, user_ptr_, event_mpi_errno_, mpi_errno_, fail_label_)	\
{														\
    mpi_errno_ = MPIDU_Socki_event_enqueue((pollinfo_), (op_), (nb_), (user_ptr_), (event_mpi_errno_));		\
    if (mpi_errno_ != MPI_SUCCESS)										\
    {														\
	mpi_errno_ = MPIR_Err_create_code(mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPIDU_SOCK_ERR_FAIL,	\
					  "**sock|poll|eqfail", "**sock|poll|eqfail %d %d %d",			\
					  pollinfo->sock_set->id, pollinfo->sock_id, (op_));			\
	goto fail_label_;											\
    }														\
}



#if (MPICH_THREAD_LEVEL != MPI_THREAD_MULTIPLE)
#   define MPIDU_SOCKI_POLLFD_OP_SET(pollfd_, pollinfo_, op_)	\
    {								\
        (pollfd_)->events |= (op_);				\
        (pollfd_)->fd = (pollinfo_)->fd;			\
    }
#   define MPIDU_SOCKI_POLLFD_OP_CLEAR(pollfd_, pollinfo_, op_)	\
    {								\
        (pollfd_)->events &= ~(op_);				\
        (pollfd_)->revents &= ~(op_);				\
        if (((pollfd_)->events & (POLLIN | POLLOUT)) == 0)	\
        {							\
            (pollfd_)->fd = -1;					\
        }							\
    }
#else
#   define MPIDU_SOCKI_POLLFD_OP_SET(pollfd_, pollinfo_, op_)		\
    {									\
	(pollinfo_)->pollfd_events |= (op_);				\
	if ((pollinfo_)->sock_set->pollfds_active == NULL)		\
	{								\
	    (pollfd_)->events |= (op_);					\
	    (pollfd_)->fd = (pollinfo_)->fd;				\
	}								\
	else								\
	{								\
	    (pollinfo_)->sock_set->pollfds_updated = TRUE;		\
	    MPIDU_Socki_wakeup((pollinfo_)->sock_set);			\
	}								\
    }
#   define MPIDU_SOCKI_POLLFD_OP_CLEAR(pollfd_, pollinfo_, op_)		\
    {									\
	(pollinfo_)->pollfd_events &= ~(op_);				\
	if ((pollinfo_)->sock_set->pollfds_active == NULL)		\
	{								\
	    (pollfd_)->events &= ~(op_);				\
	    (pollfd_)->revents &= ~(op_);				\
	    if (((pollfd_)->events & (POLLIN | POLLOUT)) == 0)		\
	    {								\
		(pollfd_)->fd = -1;					\
	    }								\
	}								\
	else								\
	{								\
	    (pollinfo_)->sock_set->pollfds_updated = TRUE;		\
	    MPIDU_Socki_wakeup((pollinfo_)->sock_set);			\
	}								\
    }
#endif

#define MPIDU_SOCKI_POLLFD_OP_ISSET(pollfd_, pollinfo_, op_) ((pollfd_)->events & (op_))

/* FIXME: Low usage operations like this should be a function for
   better readability, modularity, and code size */
#define MPIDU_SOCKI_GET_SOCKET_ERROR(pollinfo_, os_errno_, mpi_errno_, fail_label_)				\
{														\
    int rc__;													\
    socklen_t sz__;												\
														\
    sz__ = sizeof(os_errno_);											\
    rc__ = getsockopt((pollinfo_)->fd, SOL_SOCKET, SO_ERROR, &(os_errno_), &sz__);				\
    if (rc__ != 0)												\
    {														\
	if (errno == ENOMEM || errno == ENOBUFS)								\
	{													\
	    mpi_errno_ = MPIR_Err_create_code(									\
		MPI_SUCCESS, MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_NOMEM, "**sock|osnomem",	\
		"**sock|osnomem %s %d %d", "getsockopt", pollinfo->sock_set->id, pollinfo->sock_id);		\
	}													\
	else													\
	{													\
	    mpi_errno = MPIR_Err_create_code(									\
		MPI_SUCCESS, MPIR_ERR_FATAL, FCNAME, __LINE__, MPIDU_SOCK_ERR_FAIL, "**sock|oserror",		\
		"**sock|poll|oserror %s %d %d %d %s", "getsockopt", pollinfo->sock_set->id, pollinfo->sock_id,	\
		 (os_errno_), MPIU_Strerror(os_errno_));							\
	}													\
														\
        goto fail_label_;											\
    }														\
}


/*
 * Validation tests
 */
#define MPIDU_SOCKI_VERIFY_INIT(mpi_errno_, fail_label_)								\
{															\
    if (MPIDU_Socki_initialized <= 0)											\
    {															\
	(mpi_errno_) = MPIR_Err_create_code((mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_INIT,	\
					 "**sock|uninit", NULL);							\
	goto fail_label_;												\
    }															\
}


#define MPIDU_SOCKI_VALIDATE_SOCK_SET(sock_set_, mpi_errno_, fail_label_)


#define MPIDU_SOCKI_VALIDATE_SOCK(sock_, mpi_errno_, fail_label_)								\
{																\
    struct pollinfo * pollinfo__;												\
																\
    if ((sock_) == NULL || (sock_)->sock_set == NULL || (sock_)->elem < 0 ||							\
	(sock_)->elem >= (sock_)->sock_set->poll_array_elems)									\
    {																\
	(mpi_errno_) = MPIR_Err_create_code((mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_BAD_SOCK,	\
					    "**sock|badsock", NULL);								\
	goto fail_label_;													\
    }																\
																\
    pollinfo__ = MPIDU_Socki_sock_get_pollinfo(sock_);										\
																\
    if (pollinfo__->type <= MPIDU_SOCKI_TYPE_FIRST || pollinfo__->type >= MPIDU_SOCKI_TYPE_INTERRUPTER ||			\
	pollinfo__->state <= MPIDU_SOCKI_STATE_FIRST || pollinfo__->state >= MPIDU_SOCKI_STATE_LAST)				\
    {																\
	(mpi_errno_) = MPIR_Err_create_code((mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_BAD_SOCK,	\
					    "**sock|badsock", NULL);								\
	goto fail_label_;													\
    }																\
}


#define MPIDU_SOCKI_VERIFY_CONNECTED_READABLE(pollinfo_, mpi_errno_, fail_label_)						\
{																\
    if ((pollinfo_)->type == MPIDU_SOCKI_TYPE_COMMUNICATION)									\
    {																\
	if ((pollinfo_)->state == MPIDU_SOCKI_STATE_CONNECTING)									\
	{															\
	    (mpi_errno_) = MPIR_Err_create_code(										\
		(mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_BAD_SOCK, "**sock|notconnected",		\
		"**sock|notconnected %d %d", (pollinfo_)->sock_set->id, (pollinfo_)->sock_id);					\
	    goto fail_label_;													\
	}															\
	else if ((pollinfo_)->state == MPIDU_SOCKI_STATE_DISCONNECTED)								\
	{															\
	    if ((pollinfo_)->os_errno == 0)											\
	    {															\
		(mpi_errno_) = MPIR_Err_create_code(										\
		    (mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_CONN_CLOSED, "**sock|connclosed",	\
		    "**sock|connclosed %d %d", (pollinfo_)->sock_set->id, (pollinfo_)->sock_id);				\
	    }															\
	    else														\
	    {															\
		(mpi_errno_) = MPIR_Err_create_code(										\
		    (mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_CONN_FAILED, "**sock|connfailed",	\
		    "**sock|poll|connfailed %d %d %d %s", (pollinfo_)->sock_set->id, (pollinfo_)->sock_id,			\
		    (pollinfo_)->os_errno, MPIU_Strerror((pollinfo_)->os_errno));						\
	    }															\
	    goto fail_label_;													\
	}															\
	else if ((pollinfo_)->state == MPIDU_SOCKI_STATE_CLOSING)								\
	{															\
	    (mpi_errno_) = MPIR_Err_create_code(										\
		(mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_INPROGRESS, "**sock|closing",		\
		"**sock|closing %d %d", (pollinfo_)->sock_set->id, (pollinfo_)->sock_id);					\
																\
	    goto fail_label_;													\
	}															\
	else if ((pollinfo_)->state != MPIDU_SOCKI_STATE_CONNECTED_RW && (pollinfo_)->state != MPIDU_SOCKI_STATE_CONNECTED_RO)	\
	{															\
	    (mpi_errno_) = MPIR_Err_create_code((mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_BAD_SOCK,	\
						"**sock|badsock", NULL);							\
	    goto fail_label_;													\
	}															\
    }																\
    else if ((pollinfo_)->type == MPIDU_SOCKI_TYPE_LISTENER)									\
    {																\
	(mpi_errno_) = MPIR_Err_create_code(											\
	    (mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_BAD_SOCK, "**sock|listener_read",		\
	    "**sock|listener_read %d %d", (pollinfo_)->sock_set->id, (pollinfo_)->sock_id);					\
																\
	goto fail_label_;													\
    }																\
}


#define MPIDU_SOCKI_VERIFY_CONNECTED_WRITABLE(pollinfo_, mpi_errno_, fail_label_)						 \
{																 \
    if ((pollinfo_)->type == MPIDU_SOCKI_TYPE_COMMUNICATION)									 \
    {																 \
	if ((pollinfo_)->state == MPIDU_SOCKI_STATE_CONNECTING)									 \
	{															 \
	    (mpi_errno_) = MPIR_Err_create_code((mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_BAD_SOCK,	 \
						"**sock|notconnected", "**sock|notconnected %d %d",				 \
						(pollinfo_)->sock_set->id, (pollinfo_)->sock_id);				 \
	    goto fail_label_;													 \
	}															 \
	else if ((pollinfo_)->state == MPIDU_SOCKI_STATE_DISCONNECTED || (pollinfo_)->state == MPIDU_SOCKI_STATE_CONNECTED_RO)	 \
	{															 \
	    if ((pollinfo_)->os_errno == 0)											 \
	    {															 \
		(mpi_errno_) = MPIR_Err_create_code(										 \
		    (mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_CONN_CLOSED, "**sock|connclosed",	 \
		    "**sock|connclosed %d %d", (pollinfo_)->sock_set->id, (pollinfo_)->sock_id);				 \
	    }															 \
	    else														 \
	    {															 \
		(mpi_errno_) = MPIR_Err_create_code(										 \
		    (mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_CONN_FAILED, "**sock|connfailed",	 \
		    "**sock|poll|connfailed %d %d %d %s", (pollinfo_)->sock_set->id, (pollinfo_)->sock_id,			 \
		    (pollinfo_)->os_errno, MPIU_Strerror((pollinfo_)->os_errno));						 \
	    }															 \
	    goto fail_label_;													 \
	}															 \
	else if ((pollinfo_)->state == MPIDU_SOCKI_STATE_CLOSING)								 \
	{															 \
	    (mpi_errno_) = MPIR_Err_create_code((mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_INPROGRESS, \
						"**sock|closing", "**sock|closing %d %d",					 \
						(pollinfo_)->sock_set->id, (pollinfo_)->sock_id);				 \
																 \
	    goto fail_label_;													 \
	}															 \
	else if ((pollinfo_)->state != MPIDU_SOCKI_STATE_CONNECTED_RW)								 \
	{															 \
	    (mpi_errno_) = MPIR_Err_create_code((mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_BAD_SOCK,	 \
						"**sock|badsock", NULL);							 \
	    goto fail_label_;													 \
	}															 \
    }																 \
    else if ((pollinfo_)->type == MPIDU_SOCKI_TYPE_LISTENER)									 \
    {																 \
	(mpi_errno_) = MPIR_Err_create_code((mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_BAD_SOCK,	 \
					    "**sock|listener_write", "**sock|listener_write %d %d",				 \
					    (pollinfo_)->sock_set->id, (pollinfo_)->sock_id);					 \
																 \
	goto fail_label_;													 \
    }																 \
}


#define MPIDU_SOCKI_VALIDATE_FD(pollinfo_, mpi_errno_, fail_label_)								\
{																\
    if ((pollinfo_)->fd < 0)													\
    {																\
	(mpi_errno_) = MPIR_Err_create_code((mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_BAD_SOCK,	\
					    "**sock|badhandle", "**sock|poll|badhandle %d %d %d",				\
					    (pollinfo_)->sock_set->id, (pollinfo_)->sock_id, (pollinfo_)->fd);			\
	goto fail_label_;													\
    }																\
}


#define MPIDU_SOCKI_VERIFY_NO_POSTED_READ(pollfd_, pollinfo_, mpi_errno_, fail_label_)						\
{																\
    if (MPIDU_SOCKI_POLLFD_OP_ISSET((pollfd_), (pollinfo_), POLLIN))								\
    {																\
	(mpi_errno_) = MPIR_Err_create_code((mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_INPROGRESS,	\
					    "**sock|reads", "**sock|reads %d %d",						\
					    (pollinfo_)->sock_set->id, (pollinfo_)->sock_id);					\
	goto fail_label_;													\
    }																\
}


#define MPIDU_SOCKI_VERIFY_NO_POSTED_WRITE(pollfd_, pollinfo_, mpi_errno_, fail_label_)						\
{																\
    if (MPIDU_SOCKI_POLLFD_OP_ISSET((pollfd_), (pollinfo_), POLLOUT))							\
    {																\
	(mpi_errno_) = MPIR_Err_create_code((mpi_errno_), MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_INPROGRESS,	\
					    "**sock|writes", "**sock|writes %d %d",						\
					    (pollinfo_)->sock_set->id, (pollinfo_)->sock_id);					\
	goto fail_label_;													\
    }																\
}


#if (MPICH_THREAD_LEVEL == MPI_THREAD_MULTIPLE)

/*
 * MPIDU_Socki_wakeup()
 */
#undef FUNCNAME
#define FUNCNAME MPIDU_Socki_wakeup
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
static int MPIDU_Socki_wakeup(struct MPIDU_Sock_set * sock_set)
{
    if (sock_set->wakeup_posted == FALSE)
    {
	for(;;)
	{
	    int nb;
	    char c = 0;
	    
	    nb = write(sock_set->intr_fds[1], &c, 1);
	    if (nb == 1)
	    {
		break;
	    }

	    MPIU_Assertp(nb == 0 || errno == EINTR);
	}
	
	sock_set->wakeup_posted = TRUE;
    }

    return MPIDU_SOCK_SUCCESS;
}
/* end MPIDU_Socki_wakeup() */

#endif /* (MPICH_THREAD_LEVEL == MPI_THREAD_MULTIPLE) */
    

/*
 * MPIDU_Socki_os_to_mpi_errno()
 *
 * This routine assumes that no thread can change the state between state check before the nonblocking OS operation and the call
 * to this routine.
 */
#undef FUNCNAME
#define FUNCNAME MPIDU_Socki_os_to_mpi_errno
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
/* --BEGIN ERROR HANDLING-- */
static int MPIDU_Socki_os_to_mpi_errno(struct pollinfo * pollinfo, int os_errno, char * fcname, int line, int * disconnected)
{
    int mpi_errno;

    if (os_errno == ENOMEM || os_errno == ENOBUFS)
    {
	mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_RECOVERABLE, fcname, line, MPIDU_SOCK_ERR_NOMEM,
					 "**sock|osnomem", "**sock|poll|osnomem %d %d %d %s",
					 pollinfo->sock_set->id, pollinfo->sock_id, os_errno, MPIU_Strerror(os_errno));
	*disconnected = FALSE;
    }
    else if (os_errno == EFAULT || os_errno == EINVAL)
    {
	mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_RECOVERABLE, fcname, line, MPIDU_SOCK_ERR_BAD_BUF,
					 "**sock|badbuf", "**sock|poll|badbuf %d %d %d %s",
					 pollinfo->sock_set->id, pollinfo->sock_id, os_errno, MPIU_Strerror(os_errno));
	*disconnected = FALSE;
    }
    else if (os_errno == EPIPE)
    {
	mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_RECOVERABLE, fcname, line, MPIDU_SOCK_ERR_CONN_CLOSED,
					 "**sock|connclosed", "**sock|poll|connclosed %d %d %d %s",
					 pollinfo->sock_set->id, pollinfo->sock_id, os_errno, MPIU_Strerror(os_errno));
	*disconnected = TRUE;
    }
    else if (os_errno == ECONNRESET || os_errno == ENOTCONN || os_errno == ETIMEDOUT)
    {
	mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_RECOVERABLE, fcname, line, MPIDU_SOCK_ERR_CONN_FAILED,
					 "**sock|connfailed", "**sock|poll|connfailed %d %d %d %s",
					 pollinfo->sock_set->id, pollinfo->sock_id, os_errno, MPIU_Strerror(os_errno));
	pollinfo->os_errno = os_errno;
	*disconnected = TRUE;
    }
    else if (os_errno == EBADF)
    {
	/*
	 * If we have a bad file descriptor, then either the sock was bad to start with and we didn't catch it in the preliminary
	 * checks, or a sock closure was finalized after the preliminary checks were performed.  The latter should not happen if
	 * the thread safety code is correctly implemented.  In any case, the data structures associated with the sock are no
	 * longer valid and should not be modified.  We indicate this by returning a fatal error.
	 */
	mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_FATAL, fcname, line, MPIDU_SOCK_ERR_BAD_SOCK,
					 "**sock|badsock", NULL);
	*disconnected = FALSE;
    }
    else
    {
	/*
	 * Unexpected OS error.
	 *
	 * FIXME: technically we should never reach this section of code.  What's the right way to handle this situation?  Should
	 * we print an immediate message asking the user to report the errno so that we can plug the hole?
	 */
	mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_FATAL, fcname, line, MPIDU_SOCK_ERR_CONN_FAILED,
					 "**sock|oserror", "**sock|poll|oserror %d %d %d %s",
					 pollinfo->sock_set->id, pollinfo->sock_id, os_errno, MPIU_Strerror(os_errno));
	pollinfo->os_errno = os_errno;
	*disconnected = TRUE;
    }

    return mpi_errno;
}
/* --END ERROR HANDLING-- */
/* end MPIDU_Socki_os_to_mpi_errno() */