bsockets.c

MPICH是MPI的重要研究,提供了一系列的接口函数,为并行计算的实现提供了编程环境.

	    {
		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;
		i = select(maxfds, NULL, &writebfds->set, NULL, tv);
		if (i != SOCKET_ERROR)
		    nbfds += i;
	    }
	    return nbfds;
	}
    }

    if (maxfds)
	maxfds = ((BFD_Buffer*)maxfds)->real_fd + 1;
    nbfds = select(maxfds, 
	readbfds ? &readbfds->set : NULL, 
	writebfds ? &writebfds->set : NULL, 
	execbfds ? &execbfds->set : NULL, tv);
    if (nbfds == SOCKET_ERROR) 
	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++;
	    }
	}
    }
    
    return nbfds;
}

/*@
bwrite - write

  Parameters:
  +  int bfd - bsocket
  .  char *ubuf - buffer
  -  int len - length
  
    Notes:
@*/
int bwrite(int bfd, char *ubuf, int len)
{
    /*dbg_printf("bwrite\n");*/
    return bfd_write(((BFD_Buffer*)bfd)->real_fd, ubuf, len);
    /*return bfd_write(((BFD_Buffer*)bfd)->real_fd, ubuf, BSOCKET_MIN(len, 20*1024));*/
}

/*
#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;
    if (n == 0)
	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)
	{
	    return SOCKET_ERROR;
	}
    }
    DBG_BWRITEV_PRINT(("->%d)", dwNumSent));
    return dwNumSent;
#else
    int nWritten;
    /*bg_printf("bwritev\n");*/
    nWritten = writev(((BFD_Buffer*)bfd)->real_fd, pIOVec, n);
    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;
    
    DBG_MSG("Enter bread\n");
    /*dbg_printf("bread\n");*/
    
    pbfd = (BFD_Buffer*)bfd;

    if (pbfd->state == BFD_ERROR) 
	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));
	
	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) 
    {
	n = bfd_read(fd, ubuf, len);
	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;

	return n;
    }
    
    num_copied = pbfd->num_avail;
    n = bfd_read(fd, bbuf, g_bbuflen);
    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)
	    return pbfd->num_avail;
	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;
    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)
{
    int k;
    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;

    DBG_MSG("Enter breadv\n");
    /*dbg_printf("breadv\n");*/
    
    pbfd = (BFD_Buffer*)bfd;
    
    if (pbfd->state == BFD_ERROR) 
	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));
			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 += 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));
	    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
    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);
	    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);
	    n = 0; /* Set this to zero so it can be added to num_read */
	}
    }
#else
    n = readv(fd, &vec[i], veclen - i + 1);
    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));
		    return num_read;
		}
	    }
	}
    }

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

/*@
   bclose - close

   Parameters:
.  int bfd - bsocket

   Notes:
@*/
int bclose(int bfd)
{
    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 );

    return 0;
}

/*@
bgetsockname - 

  Parameters:
  +  int bfd
  .  struct sockaddr *name
  -  int *namelen
  
    Notes:
@*/
int bgetsockname(int bfd, struct sockaddr *name, int *namelen)
{
    /*dbg_printf("bgetsockname\n");*/
    return getsockname(((BFD_Buffer*)bfd)->real_fd, name, namelen);
}

/*@
make_nonblocking - make a bsocket non-blocking

  Parameters:
  . int bfd - bsocket
  
    Notes:
@*/
int bmake_nonblocking(int bfd)
{
    
    int      flag = 1;
    int      rc;
    
    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
    
    return rc;
}

/*@
make_blocking - make a bsocket blocking

  Parameters:
  . int bfd - bsocket
  
    Notes:
@*/
int bmake_blocking(int bfd)
{
    int      flag = 0;
    int      rc;
    
    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
    
    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)
{
    struct sockaddr_in sin;
    int optval = 1;
    struct linger linger;
    int len;

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

    /* Create a new bsocket */
    *bfd = bsocket(AF_INET, SOCK_STREAM, 0);
    if (*bfd == BFD_INVALID_SOCKET)
    {
	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((u_short)port);

    /* bind it to the port provided */
    if (bbind(*bfd, (const struct sockaddr *)&sin, sizeof(struct sockaddr)) == SOCKET_ERROR)
    {
	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
    return 0;
}

/*@
   beasy_connect_quick - connect without retries

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

   Notes:
@*/
int beasy_connect_quick(int bfd, char *host, int port)
{
    struct hostent *lphost;
    struct sockaddr_in sockAddr;
#ifdef USE_LINGER_SOCKOPT
    struct linger linger;
#endif
    memset(&sockAddr,0,sizeof(sockAddr));
    
    sockAddr.sin_family = AF_INET;
    sockAddr.sin_addr.s_addr = inet_addr(host);
    
    if (sockAddr.sin_addr.s_addr == INADDR_NONE || sockAddr.sin_addr.s_addr == 0)
    {
	lphost = gethostbyname(host);
	if (lphost != NULL)
	    sockAddr.sin_addr.s_addr = ((struct in_addr *)lphost->h_addr)->s_addr;
	else
	    return SOCKET_ERROR;
    }
    
    sockAddr.sin_port = htons((u_short)port);
    
    if (bconnect(bfd, (SOCKADDR*)&sockAddr, sizeof(sockAddr)) == SOCKET_ERROR)
    {
	return SOCKET_ERROR;
    }

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

    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)
{
    int error;
    int reps = 0;
    struct hostent *lphost;
    struct sockaddr_in sockAddr;
    struct linger linger;
#ifdef HAVE_WINSOCK2_H
    /* use this array to make sure the warning only gets logged once */
    BOOL bWarningLogged[4] = { FALSE, FALSE, FALSE, FALSE };
#endif

    dbg_printf("beasy_connect(%s:%d)\n", host, port);

    memset(&sockAddr,0,sizeof(sockAddr));
    
    sockAddr.sin_family = AF_INET;
    sockAddr.sin_addr.s_addr = inet_addr(host);
    
    if (sockAddr.sin_addr.s_addr == INADDR_NONE || sockAddr.sin_addr.s_addr == 0)
    {
	lphost = gethostbyname(host);
	if (lphost != NULL)
	    sockAddr.sin_addr.s_addr = ((struct in_addr *)lphost->h_addr)->s_addr;
	else
	    return SOCKET_ERROR;
    }
    
    sockAddr.sin_port = htons((u_short)port);
    
    while (bconnect(bfd, (struct sockaddr*)&sockAddr, sizeof(sockAddr)) == SOCKET_ERROR)
    {
#ifdef HAVE_WINSOCK2_H
	error = WSAGetLastError();
	srand(clock());
	if( (error == WSAECONNREFUSED || error == WSAETIMEDOUT || error == WSAENETUNREACH || error == WSAEADDRINUSE)
	    && (reps < g_beasy_connection_attempts) )
	{
	    double d = (double)rand() / (double)RAND_MAX;
	    Sleep(200 + (int)(d*200));
	    reps++;
	    switch (error)
	    {
	    case WSAECONNREFUSED:
		if (!bWarningLogged[0])
		{
		    /*log_warning("WSAECONNREFUSED error, re-attempting bconnect(%s)", host);*/
		    bWarningLogged[0] = TRUE;
		}
		break;
	    case WSAETIMEDOUT:
		if (!bWarningLogged[1])
		{
		    log_warning("WSAETIMEDOUT error, re-attempting bconnect(%s)", host);
		    bWarningLogged[1] = TRUE;
		}
		break;
	    case WSAENETUNREACH:
		if (!bWarningLogged[2])
		{
		    log_warning("WSAENETUNREACH error, re-attempting bconnect(%s)", host);
		    bWarningLogged[2] = TRUE;
		}
		break;
	    case WSAEADDRINUSE:
		if (!bWarningLogged[3])
		{
		    log_warning("WSAEADDRINUSE error, re-attempting bconnect(%s)", host);
		    bWarningLogged[3] = TRUE;
		}
		break;
	    default:
		log_warning("%d error, re-attempting bconnect");