bsockets.c

刚才是说明 现在是安装程序在 LINUX环境下进行编程的MPICH安装文件

    MPIDI_FUNC_ENTER(MPID_STATE_BCONNECT);
    /*dbg_printf("bconnect\n");*/
    ret_val = connect(((BFD_Buffer*)bfd)->real_fd, servaddr, servaddr_len);
    MPIDI_FUNC_EXIT(MPID_STATE_BCONNECT);
    return ret_val;
}

/*@
bselect - select

  Parameters:
  +  int maxfds - max bfd - 1 You must use BFD_MAX to get this value
  .  bfd_set *readbfds - read set
  .  bfd_set *writebfds - write set
  .  bfd_set *execbfds - exec set
  -  struct timeval *tv - timeout
  
    Notes:
@*/
int bselect(int maxfds, bfd_set *readbfds, bfd_set *writebfds,		   
	    bfd_set *execbfds, struct timeval *tv)
{
    int 	   nbfds;
    bfd_set        rcopy;
    BFD_Buffer     *p;
    int            i;
    MPIDI_STATE_DECL(MPID_STATE_BSELECT);
    MPIDI_STATE_DECL(MPID_STATE_SELECT);

    MPIDI_FUNC_ENTER(MPID_STATE_BSELECT);
    DBG_MSG("Enter bselect\n");
    /*dbg_printf("bselect\n");*/
    
    if (readbfds)
    {
	nbfds = 0;
	rcopy = *readbfds;
	/* check to see if there are any bfds with buffered data */
	for (i=0; i<readbfds->n; i++)
	{
	    p = readbfds->p[i];
	    if ((p->num_avail > 0) && (FD_ISSET(p->real_fd, &rcopy.set)))
	    {
		FD_SET((unsigned int)p->real_fd, &readbfds->set);
		nbfds++;
	    }
	}
	if (nbfds)
	{
	    /* buffered data is available, return it plus any writeable bfds */
	    if (writebfds)
	    {
		maxfds = ((BFD_Buffer*)maxfds)->real_fd + 1;
		MPIDI_FUNC_ENTER(MPID_STATE_SELECT);
		i = select(maxfds, NULL, &writebfds->set, NULL, tv);
		MPIDI_FUNC_EXIT(MPID_STATE_SELECT);
		if (i != SOCKET_ERROR)
		    nbfds += i;
	    }
	    MPIDI_FUNC_EXIT(MPID_STATE_BSELECT);
	    return nbfds;
	}
    }

    maxfds = ((BFD_Buffer*)maxfds)->real_fd + 1;
    MPIDI_FUNC_ENTER(MPID_STATE_SELECT);
    nbfds = select(maxfds, 
	readbfds ? &readbfds->set : NULL, 
	writebfds ? &writebfds->set : NULL, 
	execbfds ? &execbfds->set : NULL, tv);
    MPIDI_FUNC_EXIT(MPID_STATE_SELECT);
    if (nbfds == SOCKET_ERROR)
    {
	MPIDI_FUNC_EXIT(MPID_STATE_BSELECT);
	return SOCKET_ERROR;
    }
    
    if (readbfds)
    {
	for (i=0; i<readbfds->n; i++)
	{
	    p = readbfds->p[i];
	    if ((p->num_avail > 0) && (FD_ISSET(p->real_fd, &rcopy.set)) && (!(FD_ISSET(p->real_fd, &readbfds->set))))
	    {
		FD_SET((unsigned int)p->real_fd, &readbfds->set);
		nbfds++;
	    }
	}
    }

    MPIDI_FUNC_EXIT(MPID_STATE_BSELECT);
    return nbfds;
}

/*@
bwrite - write

  Parameters:
  +  int bfd - bsocket
  .  char *ubuf - buffer
  -  int len - length
  
    Notes:
@*/
int bwrite(int bfd, char *ubuf, int len)
{
    int num_written;
    MPIDI_STATE_DECL(MPID_STATE_BWRITE);

    MPIDI_FUNC_ENTER(MPID_STATE_BWRITE);
    /*dbg_printf("bwrite\n");*/
    num_written = bfd_write(((BFD_Buffer*)bfd)->real_fd, ubuf, len);
    /*return bfd_write(((BFD_Buffer*)bfd)->real_fd, ubuf, BSOCKET_MIN(len, 20*1024));*/
    MPIDI_FUNC_EXIT(MPID_STATE_BWRITE);
    return num_written;
}

/*
#define DBG_BWRITEV
#define DBG_BWRITEV_PRINT(a) BPRINTF a
*/
#undef DBG_BWRITEV
#define DBG_BWRITEV_PRINT

/*@
   bwritev - writev

   Parameters:
+  int bfd - bsocket
.  B_VECTOR *pIOVec - iovec structure
-  int n - length of iovec

   Notes:
@*/
int bwritev(int bfd, B_VECTOR *pIOVec, int n)
{
#ifdef HAVE_WINSOCK2_H
#ifdef DBG_BWRITEV
    int i;
#endif
    DWORD dwNumSent = 0;
    MPIDI_STATE_DECL(MPID_STATE_BWRITEV);

    MPIDI_FUNC_ENTER(MPID_STATE_BWRITEV);
    if (n == 0)
    {
	MPIDI_FUNC_EXIT(MPID_STATE_BWRITEV);
	return 0;
    }
#ifdef DBG_BWRITEV
    BPRINTF("(bwritev");
    for (i=0; i<n; i++)
	BPRINTF(":%d", pIOVec[i].B_VECTOR_LEN);
#endif
    if (WSASend(((BFD_Buffer*)bfd)->real_fd, pIOVec, n, &dwNumSent, 0, NULL/*overlapped*/, NULL/*completion routine*/) == SOCKET_ERROR)
    {
	if (WSAGetLastError() != WSAEWOULDBLOCK)
	{
	    MPIDI_FUNC_EXIT(MPID_STATE_BWRITEV);
	    return SOCKET_ERROR;
	}
    }
    DBG_BWRITEV_PRINT(("->%d)", dwNumSent));
    MPIDI_FUNC_EXIT(MPID_STATE_BWRITEV);
    return dwNumSent;
#else
    int nWritten;
    MPIDI_STATE_DECL(MPID_STATE_BWRITEV);

    MPIDI_FUNC_ENTER(MPID_STATE_BWRITEV);
    /*bg_printf("bwritev\n");*/
    nWritten = writev(((BFD_Buffer*)bfd)->real_fd, pIOVec, n);
    MPIDI_FUNC_EXIT(MPID_STATE_BWRITEV);
    return nWritten;
#endif
}

/*@
bread - read

  Parameters:
  +  int bfd - bsocket
  .  char *ubuf - buffer
  -  int len - length
  
    Notes:
@*/
int bread(int bfd, char *ubuf, int len)
{
    int      fd;
    int      num_used;
    int      num_copied;
    int      n;
    char     *bbuf;
    BFD_Buffer *pbfd;
    MPIDI_STATE_DECL(MPID_STATE_BREAD);
    MPIDI_STATE_DECL(MPID_STATE_READ);

    MPIDI_FUNC_ENTER(MPID_STATE_BREAD);
    DBG_MSG("Enter bread\n");
    /*dbg_printf("bread\n");*/
    
    pbfd = (BFD_Buffer*)bfd;

    if (pbfd->state == BFD_ERROR)
    {
	MPIDI_FUNC_EXIT(MPID_STATE_BREAD);
	return pbfd->errval;
    }
    
    pbfd->state = BFD_READING;
    fd = pbfd->real_fd;
    bbuf = pbfd->read_buf;
    
    if (len <= pbfd->num_avail) 
    {
	memcpy(ubuf, bbuf + pbfd->curpos, len);
	pbfd->curpos += len;
	pbfd->num_avail -= len;
	if (pbfd->num_avail == 0)
	    pbfd->curpos = 0;

	DBG_MSG(("bread: copied %d bytes into ubuf starting at bbuf[%d]\n", len, *(bbuf) + conn->curpos));

	MPIDI_FUNC_EXIT(MPID_STATE_BREAD);
	return len;
    }
    
    if (pbfd->num_avail > 0) 
    {
	memcpy(ubuf, bbuf + pbfd->curpos, pbfd->num_avail);
	ubuf += pbfd->num_avail;
	len -= pbfd->num_avail;
	pbfd->curpos = 0;
    }
    
    if (len > g_bbuflen) 
    {
	MPIDI_FUNC_ENTER(MPID_STATE_READ);
	n = bfd_read(fd, ubuf, len);
	MPIDI_FUNC_EXIT(MPID_STATE_READ);
	if (n == 0) 
	{
	    pbfd->state = BFD_ERROR;
	    pbfd->errval = 0;
	}
	else if (n == SOCKET_ERROR) 
	{
	    if ((errno != EINTR) || (errno != EAGAIN)) 
	    {
		pbfd->state = BFD_ERROR;
		pbfd->errval = errno;
	    }
	    n = 0;
	}
	
	DBG_MSG(("bread: Read %d bytes directly into ubuf\n", n));
	n += pbfd->num_avail;
	pbfd->num_avail = 0;

	MPIDI_FUNC_EXIT(MPID_STATE_BREAD);
	return n;
    }
    
    num_copied = pbfd->num_avail;
    MPIDI_FUNC_ENTER(MPID_STATE_READ);
    n = bfd_read(fd, bbuf, g_bbuflen);
    MPIDI_FUNC_EXIT(MPID_STATE_READ);
    pbfd->curpos = 0;
    if (n == 0) 
    {
	pbfd->state = BFD_ERROR;
	pbfd->errval = 0;
    }
    else if (n == SOCKET_ERROR) 
    {
	if ((errno != EINTR) || (errno != EAGAIN)) 
	{
	    pbfd->state = BFD_ERROR;
	    pbfd->errval = errno;
	}
	if (pbfd->num_avail)
	{
	    MPIDI_FUNC_EXIT(MPID_STATE_BREAD);
	    return pbfd->num_avail;
	}
	MPIDI_FUNC_EXIT(MPID_STATE_BREAD);
	return SOCKET_ERROR;
    }
    
    pbfd->num_avail = n;
    num_used = (((len) < (pbfd->num_avail)) ? (len) : (pbfd->num_avail));
    memcpy(ubuf, bbuf, num_used);
    pbfd->curpos += num_used;
    pbfd->num_avail -= num_used;
    /*if (pbfd->num_avail > 0) BPRINTF("bread: %d extra bytes read into bbuf %d\n", pbfd->num_avail, pbfd->real_fd);*/
    pbfd->state = BFD_IDLE;

    DBG_MSG(("bread: Read %d bytes on socket %d into bbuf\n", n, fd));
    DBG_MSG(("bread: copied %d bytes into ubuf from bbuf\n", num_used));
    
    n = num_used + num_copied;
    MPIDI_FUNC_EXIT(MPID_STATE_BREAD);
    return n;
}

/*
#define DBG_BREADV
#define DBG_BREADV_PRINT(a) BPRINTF a
*/
#undef DBG_BREADV
#define DBG_BREADV_PRINT(a) 

/*@
   breadv - readv

   Parameters:
+  int bfd - bsocket
.  B_VECTOR *uvec - iovec array
-  int len - length of array

   Notes:
   The vec parameter must have one more element than veclen.  This extra
   element is used by this function to read additional data into an internal
   buffer.
   The elements of the vec parameter may be changed by this function.
@*/
int breadv(int bfd, B_VECTOR *vec, int veclen)
{
#ifdef DBG_READV
    int k;
#endif
    int      fd;
    int      i;
    char     *bbuf;
    BFD_Buffer *pbfd;
    int      num_read = 0;
#ifdef HAVE_WINSOCK2_H
    DWORD    n = 0;
    DWORD    nFlags = 0;
#else
    int      n = 0;
#endif
    B_VECTOR pVector[B_VECTOR_LIMIT];
    int iVector;
    MPIDI_STATE_DECL(MPID_STATE_BREADV);
    MPIDI_STATE_DECL(MPID_STATE_READV);

    MPIDI_FUNC_ENTER(MPID_STATE_BREADV);
    DBG_MSG("Enter breadv\n");
    /*dbg_printf("breadv\n");*/
    
    pbfd = (BFD_Buffer*)bfd;
    
    if (pbfd->state == BFD_ERROR)
    {
	MPIDI_FUNC_EXIT(MPID_STATE_BREADV);
	return pbfd->errval;
    }
    
    pbfd->state = BFD_READING;
    fd = pbfd->real_fd;
    bbuf = pbfd->read_buf;
    
#ifdef DBG_BREADV
    BPRINTF("(breadv");
    for (i=0; i<veclen; i++)
	BPRINTF(":%d", vec[i].B_VECTOR_LEN);
#endif
    num_read = 0;
    for (i=0; i<veclen; i++)
    {
	if (pbfd->num_avail)
	{
	    n = BSOCKET_MIN((unsigned int)pbfd->num_avail, vec[i].B_VECTOR_LEN);
	    DBG_BREADV_PRINT((",bcopy %d", n));
	    memcpy(vec[i].B_VECTOR_BUF, bbuf + pbfd->curpos, n);
	    if ((unsigned int)pbfd->num_avail <= vec[i].B_VECTOR_LEN)
	    {
		if ((unsigned int)pbfd->num_avail == vec[i].B_VECTOR_LEN)
		{
		    i++;
		    if (i==veclen)
		    {
			pbfd->num_avail = 0;
			pbfd->curpos = 0;
			DBG_BREADV_PRINT(("->%d,%da)", num_read+n, pbfd->num_avail));
			MPIDI_FUNC_EXIT(MPID_STATE_BREADV);
			return num_read + n;
		    }
		}
		else
		{
		    /* Make a copy of the vector */
		    for (iVector = 0; iVector <= i; iVector++)
		    {
			pVector[iVector].B_VECTOR_BUF = vec[iVector].B_VECTOR_BUF;
			pVector[iVector].B_VECTOR_LEN = vec[iVector].B_VECTOR_LEN;
		    }
		    pVector[i].B_VECTOR_BUF = (char*)(pVector[i].B_VECTOR_BUF) + n;
		    pVector[i].B_VECTOR_LEN -= n;
		    vec = pVector;
		}
	    }
	    pbfd->num_avail -= n;
	    pbfd->curpos += n;
	    num_read += n;
	}
	
	if (pbfd->num_avail == 0)
	{
	    pbfd->curpos = 0;
	    break;
	}
	if (i == veclen - 1)
	{
	    DBG_BREADV_PRINT(("->%d,%db)", num_read, conn->num_avail));
	    MPIDI_FUNC_EXIT(MPID_STATE_BREADV);
	    return num_read;
	}
    }
    
    vec[veclen].B_VECTOR_BUF = bbuf;
    vec[veclen].B_VECTOR_LEN = g_bbuflen;

#ifdef DBG_BREADV
    BPRINTF(",breadv");
    for (k=0; k<veclen-i+1; k++)
	BPRINTF(":%d", vec[k+i].B_VECTOR_LEN);
#endif
#ifdef HAVE_WINSOCK2_H
    MPIDI_FUNC_ENTER(MPID_STATE_READV);
    if (WSARecv(fd, &vec[i], veclen - i + 1, &n, &nFlags, NULL/*overlapped*/, NULL/*completion routine*/) == SOCKET_ERROR)
    {
	if (WSAGetLastError() != WSAEWOULDBLOCK)
	{
	    pbfd->state = BFD_ERROR;
	    pbfd->errval = WSAGetLastError();
	    BPRINTF("***WSARecv failed reading %d WSABUFs, error %d***\n", veclen - i + 1, pbfd->errval);
#ifdef DBG_READV
	    for (k=0; k<veclen-i+1; k++)
		BPRINTF("vec[%d] len: %d\nvec[%d] buf: 0x%x\n", k+i, vec[k+i].B_VECTOR_LEN, k+i, vec[k+i].B_VECTOR_BUF);
#endif
	    n = 0; /* Set this to zero so it can be added to num_read */
	}
    }
    MPIDI_FUNC_EXIT(MPID_STATE_READV);
#else
    MPIDI_FUNC_ENTER(MPID_STATE_READV);
    n = readv(fd, &vec[i], veclen - i + 1);
    MPIDI_FUNC_EXIT(MPID_STATE_READV);
    if (n == SOCKET_ERROR) 
    {
	if ((errno != EINTR) || (errno != EAGAIN)) 
	{
	    pbfd->state = BFD_ERROR;
	    pbfd->errval = errno;
	}
	n = 0; /* Set this to zero so it can be added to num_read */
    }
#endif
    
    if (n)
    {
	for ( ; i <= veclen; i++)
	{
	    if (i == veclen)
	    {
		pbfd->num_avail = n;
	    }
	    else
	    {
		num_read += BSOCKET_MIN(vec[i].B_VECTOR_LEN, n);
		n = n - BSOCKET_MIN(vec[i].B_VECTOR_LEN, n);
		if (n == 0)
		{
		    DBG_BREADV_PRINT(("->%d,%dc)", num_read, pbfd->num_avail));
		    MPIDI_FUNC_EXIT(MPID_STATE_BREADV);
		    return num_read;
		}
	    }
	}
    }

    DBG_BREADV_PRINT(("->%d,%dd)", num_read, pbfd->num_avail));
    MPIDI_FUNC_EXIT(MPID_STATE_BREADV);
    return num_read;
}

/*@
   bclose - close

   Parameters:
.  int bfd - bsocket

   Notes:
@*/
int bclose(int bfd)
{
    MPIDI_STATE_DECL(MPID_STATE_BCLOSE);

    MPIDI_FUNC_ENTER(MPID_STATE_BCLOSE);
    DBG_MSG("Enter bclose\n");
    /*dbg_printf("bclose\n");*/

    bfd_close(((BFD_Buffer*)bfd)->real_fd);
    memset((void*)bfd, 0, sizeof(BFD_Buffer));
    BlockFree( Bsocket_mem, (BFD_Buffer*)bfd );

    MPIDI_FUNC_EXIT(MPID_STATE_BCLOSE);
    return 0;
}

/*@
bgetsockname - 

  Parameters:
  +  int bfd
  .  struct sockaddr *name
  -  int *namelen
  
    Notes:
@*/
int bgetsockname(int bfd, struct sockaddr *name, int *namelen)
{
    int ret_val;
    MPIDI_STATE_DECL(MPID_STATE_BGETSOCKNAME);

    MPIDI_FUNC_ENTER(MPID_STATE_BGETSOCKNAME);
    /*dbg_printf("bgetsockname\n");*/
    ret_val = getsockname(((BFD_Buffer*)bfd)->real_fd, name, namelen);
    MPIDI_FUNC_EXIT(MPID_STATE_BGETSOCKNAME);
    return ret_val;
}

/*@
make_nonblocking - make a bsocket non-blocking

  Parameters:
  . int bfd - bsocket
  
    Notes:
@*/
int bmake_nonblocking(int bfd)
{
    
    int      flag = 1;
    int      rc;
    MPIDI_STATE_DECL(MPID_STATE_BMAKE_NONBLOCKING);

    MPIDI_FUNC_ENTER(MPID_STATE_BMAKE_NONBLOCKING);
    DBG_MSG("Enter make_nonblocking\n");
    /*dbg_printf("bmake_nonblocking\n");*/
    
#ifdef HAVE_WINDOWS_SOCKET
    rc = ioctlsocket(((BFD_Buffer*)bfd)->real_fd, FIONBIO, &flag);
#else
    rc = ioctl(((BFD_Buffer*)bfd)->real_fd, FIONBIO, &flag);
#endif
    MPIDI_FUNC_EXIT(MPID_STATE_BMAKE_NONBLOCKING);
    return rc;
}

/*@
make_blocking - make a bsocket blocking

  Parameters:
  . int bfd - bsocket
  
    Notes:
@*/
int bmake_blocking(int bfd)
{
    int      flag = 0;
    int      rc;
    MPIDI_STATE_DECL(MPID_STATE_BMAKE_BLOCKING);
    
    MPIDI_FUNC_ENTER(MPID_STATE_BMAKE_BLOCKING);
    DBG_MSG("Enter make_blocking\n");
    /*dbg_printf("bmake_blocking\n");*/
    
#ifdef HAVE_WINDOWS_SOCKET
    rc = ioctlsocket(((BFD_Buffer*)bfd)->real_fd, FIONBIO, &flag);
#else
    rc = ioctl(((BFD_Buffer*)bfd)->real_fd, FIONBIO, &flag);
#endif
    MPIDI_FUNC_EXIT(MPID_STATE_BMAKE_BLOCKING);
    return rc;
}

#endif /* NO_BSOCKETS #else */

/*@
   beasy_create - create a bsocket

   Parameters:
+  int *bfd - bsocket
.  int port - port
-  unsigned long addr - address

   Notes:
@*/
int beasy_create(int *bfd, int port, unsigned long addr)
{
#ifdef HAVE_WINSOCK2_H
    int len;
#endif
    struct sockaddr_in sin;
    int optval = 1;
    struct linger linger;
    MPIDI_STATE_DECL(MPID_STATE_BEASY_CREATE);

    MPIDI_FUNC_ENTER(MPID_STATE_BEASY_CREATE);

    /*dbg_printf("beasy_create\n");*/

    /* Create a new bsocket */
    *bfd = bsocket(AF_INET, SOCK_STREAM, 0);
    if (*bfd == BFD_INVALID_SOCKET)
    {
	MPIDI_FUNC_EXIT(MPID_STATE_BEASY_CREATE);
	return SOCKET_ERROR;
    }
    
    memset(&sin, 0, sizeof(struct sockaddr_in));
    sin.sin_family = AF_INET;
    sin.sin_addr.s_addr = addr;
    sin.sin_port = htons((unsigned short)port);

    /* bind it to the port provided */
    if (bbind(*bfd, (const struct sockaddr *)&sin, sizeof(struct sockaddr)) == SOCKET_ERROR)
    {
	MPIDI_FUNC_EXIT(MPID_STATE_BEASY_CREATE);
	return SOCKET_ERROR;
    }

    /* Set the no-delay option */
    bsetsockopt(*bfd, IPPROTO_TCP, TCP_NODELAY, (char *)&optval, sizeof(optval));

    /* Set the linger on close option */
    linger.l_onoff = 1 ;
    linger.l_linger = 60;
    bsetsockopt(*bfd, SOL_SOCKET, SO_LINGER, (char*)&linger, sizeof(linger));

#ifdef HAVE_WINSOCK2_H
    /* set the socket buffer size to 64k */
    len = sizeof(int);
    if (!getsockopt(bget_fd((*bfd)), SOL_SOCKET, SO_RCVBUF, (char*)&optval, &len))
    {
	optval = 64*1024;
	bsetsockopt(*bfd, SOL_SOCKET, SO_RCVBUF, (char*)&optval, sizeof(int));
    }
    len = sizeof(int);
    if (!getsockopt(bget_fd((*bfd)), SOL_SOCKET, SO_SNDBUF, (char*)&optval, &len))
    {
	optval = 64*1024;
	bsetsockopt(*bfd, SOL_SOCKET, SO_SNDBUF, (char*)&optval, sizeof(int));
    }
#endif
    MPIDI_FUNC_EXIT(MPID_STATE_BEASY_CREATE);
    return 0;
}

/*@
   beasy_connect - connect

   Parameters:
+  int bfd - bsocket
.  char *host - hostname
-  int port - port

   Notes:
@*/
int beasy_connect(int bfd, char *host, int port)
{
#ifdef HAVE_WINSOCK2_H
    int error;
#endif
    int reps = 0;
    struct hostent *lphost;
    struct sockaddr_in sockAddr;
    struct linger linger;