bsockets.c

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

/* -*- Mode: C; c-basic-offset:4 ; -*- */
/*
 *  (C) 2001 by Argonne National Laboratory.
 *      See COPYRIGHT in top-level directory.
 */
#include "bsocketimpl.h"
#ifdef HAVE_STDIO_H
#include <stdio.h>
#endif
#ifdef HAVE_UNISTD_H
#include <unistd.h>
#endif
#ifdef HAVE_SYS_FILIO_H
#include <sys/filio.h>
#elif defined(HAVE_SYS_IOCTL_H)
#include <sys/ioctl.h>
#endif
#include <stdio.h>
#ifdef HAVE_SYS_TIME_H
#include <sys/time.h> 
#endif
#include <errno.h> 
#ifdef HAVE_SYS_PARAM_H
#include <sys/param.h>
#endif
#ifdef HAVE_STDLIB_H
#include <stdlib.h>
#endif
#ifdef HAVE_SYS_SOCKET_H
#include <sys/socket.h>
#endif
/* FIONBIO (solaris sys/filio.h) */
#ifdef HAVE_SYS_IOCTL_H
#include <sys/ioctl.h> 
#endif
/* TCP_NODELAY */
#ifdef HAVE_NETINET_TCP_H
#include <netinet/tcp.h> 
#endif
/* defs of gethostbyname */
#ifdef HAVE_NETDB_H
#include <netdb.h> 
#endif
#ifdef HAVE_SYS_SELECT_H
#include <sys/select.h>
#endif
#ifdef HAVE_WINSOCK2_H
#include <time.h>
#endif

/*#define DEBUG_BSOCKET*/
#undef DEBUG_BSOCKET

#define BPRINTF printf

#ifdef DEBUG_BSOCKET
#define DBG_MSG(paramlist) BPRINTF( paramlist )
#else
#define DBG_MSG(paramlist) 
#endif

#define BSTRINGLEN 20

#if !defined(NO_BSOCKETS)

typedef enum { 
    BFD_FD_NOT_IN_USE, 
    BFD_ALLOCATING, 
    BFD_NEW_FD, 
    BFD_BOUND, 
    BFD_LISTENING, 
    BFD_ACCEPTED, 
    BFD_CONNECTED, 
    BFD_WRITING, 
    BFD_READING, 
    BFD_IDLE, 
    BFD_NOT_READY, 
    BFD_SOCKET_READY, 
    BFD_ERROR 
} BFD_State;

struct BFD_Buffer_struct {
    int        real_fd;        /* socket descriptor */
    int        read_flag;      /* set if reading */
    int        write_flag;     /* set if writing */
    int        curpos;         /* holds current position in bbuf */
    int        num_avail;      /* bytes in our buffered read buffer */
    BFD_State  state;          /* state of our socket */
    int        errval;         /* errno value */
    char       string[BSTRINGLEN];
    char       read_buf[1];    /* read buffer */
    /* do not add any members after the read_buf because they will be overwritten */
    /* the read_buf continues past the end of this structure by g_bbuflen bytes */
};

BlockAllocator Bsocket_mem;

#endif /* !defined(NO_BSOCKETS) */

#define BSOCKET_MIN(a, b) ((a) < (b) ? (a) : (b))
#define BSOCKET_MAX(a, b) ((a) > (b) ? (a) : (b))

static int g_beasy_connection_attempts = 5;

#ifdef HAVE_WINSOCK2_H
static void log_warning(char *str, ...)
{
    char    szMsg[256] = "bsocket error";
    HANDLE  hEventSource;
    char   *lpszStrings[2];
    char pszStr[4096];
    va_list list;

    va_start(list, str);
    vsprintf(pszStr, str, list);
    va_end(list);
    
    hEventSource = RegisterEventSource(NULL, "bsocket");
    
    lpszStrings[0] = szMsg;
    lpszStrings[1] = pszStr;
    
    if (hEventSource != NULL) 
    {
	ReportEvent(hEventSource, /* handle of event source */
	    EVENTLOG_WARNING_TYPE,  /* event type */
	    0,                    /* event category */
	    0,                    /* event ID */
	    NULL,                 /* current user's SID */
	    2,                    /* strings in lpszStrings */
	    0,                    /* no bytes of raw data */
	    (LPCTSTR*)lpszStrings,/* array of error strings */
	    NULL);                /* no raw data */
	
	DeregisterEventSource(hEventSource);
    }
}
#else
#define log_warning()
#endif

#ifdef NO_BSOCKETS

static int g_nInitRefCount = 0;
int bsocket_init(void)
{
    char *szNum;
#ifdef HAVE_WINSOCK2_H
    WSADATA wsaData;
    int err;
#endif

    if (g_nInitRefCount)
    {
	g_nInitRefCount++;
	return 0;
    }

#ifdef HAVE_WINSOCK2_H
    /* Start the Winsock dll */
    if ((err = WSAStartup(MAKEWORD(2, 0), &wsaData)) != 0)
    {
	printf("Winsock2 dll not initialized, error %d\n", err);
	return err;
    }
#endif

    szNum = getenv("BSOCKET_CONN_TRIES");
    if (szNum != NULL)
    {
	g_beasy_connection_attempts = atoi(szNum);
	if (g_beasy_connection_attempts < 1)
	    g_beasy_connection_attempts = 5;
    }

    g_nInitRefCount++;

    return 0;
}

int bsocket_finalize(void)
{
    g_nInitRefCount--;
    if (g_nInitRefCount < 1)
	g_nInitRefCount = 0;
    else
	return 0;

#ifdef HAVE_WINSOCK2_H
    WSACleanup();
#endif

    return 0;
}

char g_btempstr[BSTRINGLEN];
char *bto_string( int bfd )
{
    snprintf(g_btempstr, BSTRINGLEN, "%d", bfd);
    return g_btempstr;
}

void bprint_set(bfd_set *p)
{
    unsigned int i;
    if (p->fd_count < 1)
	return;
    for (i=0; i<p->fd_count; i++)
	printf("%d ", p->fd_array[i]);
    printf("\n");
    fflush(stdout);
}

int bwritev(int bfd, B_VECTOR *pIOVec, int n)
{
#ifdef HAVE_WINSOCK2_H
    DWORD dwNumSent = 0;
    if (n == 0)
	return 0;
    if (WSASend(bfd, pIOVec, n, &dwNumSent, 0, NULL/*overlapped*/, NULL/*completion routine*/) == SOCKET_ERROR)
    {
	if (WSAGetLastError() != WSAEWOULDBLOCK)
	{
	    return SOCKET_ERROR;
	}
    }
    return dwNumSent;
#else
    return writev(bfd, pIOVec, n);
#endif
}

int breadv(int bfd, B_VECTOR *vec, int veclen)
{
#ifdef HAVE_WINSOCK2_H
    /*int k;*/
    DWORD    n = 0;
    DWORD    nFlags = 0;
#else
    int      n = 0;
#endif

    DBG_MSG("Enter breadv\n");
    
#ifdef HAVE_WINSOCK2_H
    if (WSARecv(bfd, vec, veclen, &n, &nFlags, NULL/*overlapped*/, NULL/*completion routine*/) == SOCKET_ERROR)
    {
	if (WSAGetLastError() != WSAEWOULDBLOCK)
	{
	    /*
	    for (k=0; k<veclen; k++)
		msg_printf("vec[%d] len: %d\nvec[%d] buf: 0x%x\n", k, vec[k].B_VECTOR_LEN, k, vec[k].B_VECTOR_BUF);
	    */
	    return SOCKET_ERROR;
	}
	n = 0;
    }
#else
    n = readv(bfd, vec, veclen);
#endif
    
    return n;
}

int bmake_nonblocking(int bfd)
{
    
    int      flag = 1;
    int      rc;
    
    DBG_MSG("Enter make_nonblocking\n");
    
#ifdef HAVE_WINDOWS_SOCKET
    rc = ioctlsocket(bfd, FIONBIO, &flag);
#else
    rc = ioctl(bfd, FIONBIO, &flag);
#endif
    
    return rc;
}

int bmake_blocking(int bfd)
{
    int      flag = 0;
    int      rc;
    
    DBG_MSG("Enter make_blocking\n");
    
#ifdef HAVE_WINDOWS_SOCKET
    rc = ioctlsocket(bfd, FIONBIO, &flag);
#else
    rc = ioctl(bfd, FIONBIO, &flag);
#endif
    
    return rc;
}

#else /* #ifdef NO_BSOCKETS */

#define BBUF_LOWER_LIMIT 100
#define BBUF_DEFAULT_LEN 1024

static int g_bbuflen = BBUF_DEFAULT_LEN;
static int g_buf_pool_size = FD_SETSIZE;

/*@
   bget_fd - get fd

   Parameters:
+  int bfd - bfd

   Notes:
@*/
unsigned int bget_fd(int bfd)
{
    /*dbg_printf("bget_fd\n");*/
    return (unsigned int)(((BFD_Buffer*)bfd)->real_fd);
}

/*@
   bset - bset

   Parameters:
+  int bfd - bfd
-  bfd_set *s - set

   Notes:
@*/
void bset(int bfd, bfd_set *s)
{
    int i;
    /*dbg_printf("bset\n");*/
    FD_SET( bget_fd(bfd), & (s) -> set );
    for (i=0; i<s->n; i++)
    {
	if (s->p[i] == (BFD_Buffer*)bfd)
	    return;
    }
    s->p[s->n] = (BFD_Buffer*)bfd;
    s->n++;
}

/*@
   bclr - blcr

   Parameters:
+  int bfd - bfd
-  bfd_set *s - set

   Notes:
@*/
void bclr(int bfd, bfd_set *s)
{
    int i;
    BFD_Buffer* p;

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

    FD_CLR( bget_fd(bfd), & (s) -> set );

    if (s->n == 0)
	return;

    p = (BFD_Buffer*)bfd;
    for (i=0; i<s->n; i++)
    {
	if (s->p[i] == p)
	{
	    s->p[i] = s->p[s->n-1];
	    s->n--;
	    return;
	}
    }
}

/*@
bsocket_init - init

  
    Notes:
@*/
static int g_nInitRefCount = 0;
int bsocket_init(void)
{
    char *pszEnvVar;
    char *szNum;
#ifdef HAVE_WINSOCK2_H
    WSADATA wsaData;
    int err;
#endif
    
    if (g_nInitRefCount)
    {
	g_nInitRefCount++;
	return 0;
    }

#ifdef HAVE_WINSOCK2_H
    /* Start the Winsock dll */
    if ((err = WSAStartup(MAKEWORD(2, 0), &wsaData)) != 0)
    {
	BPRINTF("Winsock2 dll not initialized, error %d\n", err);
	return err;
    }
#endif

    szNum = getenv("BSOCKET_CONN_TRIES");
    if (szNum != NULL)
    {
	g_beasy_connection_attempts = atoi(szNum);
	if (g_beasy_connection_attempts < 1)
	    g_beasy_connection_attempts = 5;
    }

    pszEnvVar = getenv("BSOCKET_BBUFLEN");
    if (pszEnvVar != NULL)
    {
	g_bbuflen = atoi(pszEnvVar);
	if (g_bbuflen < BBUF_LOWER_LIMIT)
	    g_bbuflen = BBUF_DEFAULT_LEN;
    }

    Bsocket_mem = BlockAllocInit(sizeof(BFD_Buffer) + g_bbuflen, 64, 64, malloc, free);

    g_nInitRefCount++;

    return 0;
}

/*@
bsocket_finalize - finalize

  
    Notes:
@*/
int bsocket_finalize(void)
{
    dbg_printf("bsocket_finalize\n");
    g_nInitRefCount--;
    if (g_nInitRefCount < 1)
	g_nInitRefCount = 0;
    else
	return 0;

    /* Free up the memory used by Bsocket_mem */
    BlockAllocFinalize(&Bsocket_mem);
    
#ifdef HAVE_WINSOCK2_H
    WSACleanup();
#endif

    return 0;
}

char *bto_string( int bfd )
{
    return ((BFD_Buffer*)bfd)->string;
}

void bprint_set(bfd_set *p)
{
    unsigned int i;
    if (p->set.fd_count < 1)
	return;
    for (i=0; i<p->set.fd_count; i++)
	printf("%d ", p->set.fd_array[i]);
    printf("\n");
    fflush(stdout);
}

/*@
bsocket - socket

  Parameters:
+  int family - family
.  int type - type
-  int protocol - protocol
  
    Notes:
@*/
int bsocket(int family, int type, int protocol)
{
#ifdef HAVE_WINSOCK2_H
    int bfdtemp;
#endif
    BFD_Buffer *pbfd;

    DBG_MSG("Enter bsocket\n");
    /*dbg_printf("bsocket\n");*/
    
    pbfd = (BFD_Buffer *)BlockAlloc( Bsocket_mem );
    if (pbfd == 0) 
    {
	DBG_MSG(("ERROR in bsocket: BlockAlloc returned NULL"));
	return BFD_INVALID_SOCKET;
    }
    
    memset(pbfd, 0, sizeof(BFD_Buffer));
    pbfd->state = BFD_FD_NOT_IN_USE;

#ifdef HAVE_WINSOCK2_H
    bfdtemp = socket(family, type, protocol);
    DuplicateHandle(GetCurrentProcess(), (HANDLE)bfdtemp, GetCurrentProcess(), &(HANDLE)(pbfd->real_fd), 0, FALSE, DUPLICATE_CLOSE_SOURCE | DUPLICATE_SAME_ACCESS);
#else
    pbfd->real_fd = socket(family, type, protocol);
#endif

    if (pbfd->real_fd == SOCKET_ERROR) 
    {
	DBG_MSG("ERROR in bsocket: socket returned SOCKET_ERROR\n");
	memset(pbfd, 0, sizeof(BFD_Buffer));
	BlockFree( Bsocket_mem, pbfd );
	return BFD_INVALID_SOCKET;
    }

    snprintf(pbfd->string, BSTRINGLEN, "0x%p:%d", pbfd, pbfd->real_fd);

    return (int)pbfd;
}

/*@
bbind - bind

Parameters:
+  int bfd - bsocket
.  const struct sockaddr *servaddr - address
-  socklen_t servaddr_len - address length
  
    Notes:
@*/
int bbind(int bfd, const struct sockaddr *servaddr,	      
	  socklen_t servaddr_len)
{
    DBG_MSG("Enter bbind\n");
    /*dbg_printf("bbind\n");*/
    
    return bind(((BFD_Buffer*)bfd)->real_fd, servaddr, servaddr_len);
}

/*@
blisten - listen

Parameters:
+  int bfd - bsocket
-  int backlog - backlog
  
    Notes:
@*/
int blisten(int bfd, int backlog)
{
    /*dbg_printf("blisten\n");*/
    return listen(((BFD_Buffer*)bfd)->real_fd, backlog);
}

/*@
bsetsockopt - setsockopt

  Parameters:
  +  int bfd - bsocket
  .  int level - level
  .  int optname - optname
  .  const void *optval - optval
  -  socklen_t optlen - optlen
  
    Notes:
@*/
int bsetsockopt(int bfd, int level, int optname, const void *optval,		    
		socklen_t optlen)
{
    /*dbg_printf("bsetsockopt\n");*/
    return setsockopt(((BFD_Buffer*)bfd)->real_fd, level, optname, optval, optlen);
}

/*@
baccept - accept

  Parameters:
  +  int bfd - bsocket
  .  struct sockaddr *cliaddr - client address
  -  socklen_t *clilen - address length
  
    Notes:
@*/
int baccept(int bfd, struct sockaddr *cliaddr, socklen_t *clilen)
{
    int 	       conn_fd, bfdtemp;
    BFD_Buffer 	       *new_bfd;
    
    DBG_MSG("Enter baccept\n");
    /*dbg_printf("baccept\n");*/
    
    bfdtemp = accept(((BFD_Buffer*)bfd)->real_fd, cliaddr, clilen);
    if (bfdtemp == SOCKET_ERROR) 
    {
	DBG_MSG("ERROR in baccept: accept returned SOCKET_ERROR\n");
	return BFD_INVALID_SOCKET;
    }
    
    new_bfd = (BFD_Buffer *)BlockAlloc( Bsocket_mem );
    if (new_bfd == 0) 
    {
	DBG_MSG(("ERROR in baccept: BlockAlloc return NULL\n"));
	return BFD_INVALID_SOCKET;
    }

#ifdef HAVE_WINSOCK2_H
    DuplicateHandle(GetCurrentProcess(), (HANDLE)bfdtemp, GetCurrentProcess(), (HANDLE*)&conn_fd, 0, FALSE, DUPLICATE_CLOSE_SOURCE | DUPLICATE_SAME_ACCESS);
#else
    conn_fd = bfdtemp;
#endif

    memset(new_bfd, 0, sizeof(BFD_Buffer));
    new_bfd->real_fd = conn_fd;
    new_bfd->state = BFD_IDLE;

    snprintf(new_bfd->string, BSTRINGLEN, "0x%p:%d", new_bfd, new_bfd->real_fd);

    return (int)new_bfd;
}

/*@
bconnect - connect

  Parameters:
  +  int bfd - bsocket
  .  const struct sockaddr *servaddr - address
  -  socklen_t servaddr_len - address length
  
    Notes:
@*/
int bconnect(int bfd, const struct sockaddr *servaddr,		    
	     socklen_t servaddr_len)
{
    /*dbg_printf("bconnect\n");*/
    return connect(((BFD_Buffer*)bfd)->real_fd, servaddr, servaddr_len);
}

/*@
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;

    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)))