tcpvx.c

extremeDB s sample code,useful for you

/*******************************************************************
 *                                                                 *
 *  This file is a part of the eXtremeDB HA framework              *
 *  Copyright (c) 2001-2006 McObject LLC                           * 
 *  All Rights Reserved                                            *
 *                                                                 *
 *******************************************************************/
/*
 * ++
 * 
 * PROJECT:   eXtremeDB(tm) (c) McObject LLC
 *
 * SUBSYSTEM: HA support
 *
 * MODULE:    tcpvx.C
 *
 * ABSTRACT:  Low level OS dependent IPC.
 *            Channel-over-TCP, platform dependent functions
 *
 *
 * VERSION:   1.0
 *
 * HISTORY:
 *            1.0- 1 SS     19-Feb-2004 Created it was
 *
 * --
 */

#define INCLUDE_SOCKETS
#include "interface.h"

#ifdef CFG_TCP_SOCKET_CHANNEL

#include "ioLib.h"
#include "sigLib.h"
#include "stdio.h"

extern  int2 allocate_index();
extern  void deallocate_index(int2 index);

extern uint4        NumBytes;

int                 conn_flag = 0;
timer_unit          init_time = 0;
int                 commit_initialized = 0;

/******************************************************************************
 * function init_channel() - initializes the channel descriptor
 */
static void init_channel( nw_channel_h chan )
{
    chan->index  = -1;
    chan->status = 0;
    chan->socket        = INVALID_SOCKET_VALUE;
    chan->cancel_socket = INVALID_SOCKET_VALUE;
}

/****************************************************************************
 * Transport layer. Used by an application for providing of communication   *
 * channels, guaranteed data transfer. Can be used by an application for    *
 * the implementation of it's own communications channels.                  * 
 * Some details of a communication channel are accessible to application    * 
 ****************************************************************************/

/***************************************************************************
 *
 * Close the communication channel
 *
 * Parameters:
 *
 * IN nw_channel_h ch      - pointer to the channel descriptor.
 *
 * Description:
 * 
 * Closes the communication channel
 *
 * Return:
 * 
 * Returns MSO_S_OK if successful or error code.
 */
MCO_RET nw_close( nw_channel_h chan)
{ 
  MCO_RET rc = MCO_S_OK;

    deallocate_index(chan->index);
    if(chan->socket != INVALID_SOCKET_VALUE) {
      if(close(chan->socket)) rc = (MCO_RET)MCO_E_NW_ERROR;
    }
    if(chan->cancel_socket != INVALID_SOCKET_VALUE) {
      if(close(chan->cancel_socket)) rc = (MCO_RET)MCO_E_NW_ERROR;
    }
    init_channel(chan);
    return rc;
}

/***************************************************************************
 * 
 * Initialize the socket support
 * 
 * Parameters:
 *
 * IN nw_channel_h chan   - pointer to the base channel descriptor.
 *
 * Description:
 *
 * Initializes the network protocol & base channel descriptor.
 * The base channel is the main channel descriptor used by server for accepting
 * client's connections requests. The descriptor of base channel MAY contain
 * extention including common properties of the transport layer.
 *
 * Return:
 *
 * Returns MSO_S_OK if successful or error code.
 */
MCO_RET nw_init( nw_channel_h chan)
{
  /* The WinSock DLL is acceptable. Proceed. */
  init_channel( chan );
  chan->status = NWST_INITIALIZED;       // base channel status = initialized 

  return MCO_S_OK;
}

/**************************************************************************
 *
 * Create nonblocked TCP socket. Internal function
 *
 * Parameters:
 *
 *  SOCKET*  psock       - pointer to a socket handle variable
 * 
 * Description:
 * 
 * Creates an overlapped TCP socket.
 *
 * Return:
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
static MCO_RET nw_create_socket_internal( SOCKET*  psock )
{
  int optval = 1;
  SOCKET sock;

    if ( (sock = socket(PF_INET, SOCK_STREAM, 0)) < 0 )  {
      *psock = INVALID_SOCKET_VALUE;
      return (MCO_RET)(MCO_E_NW_FATAL);
    }
    *psock = sock;

    setsockopt (sock, IPPROTO_TCP, TCP_NODELAY,
      (const char*)&optval, sizeof(int) );
      
    setsockopt (sock, SOL_SOCKET, SO_REUSEADDR,
      (const char*)&optval, sizeof(int) );

    return MCO_S_OK;
}

/**************************************************************************
 *
 * Create nonblocked TCP socket.
 *
 * Parameters:
 *
 * IN nw_channel_h  ch      - pointer to a channel descriptor.
 * 
 * Description:
 * 
 * Creates an overlapped TCP socket.
 *
 * Return:
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
MCO_RET nw_create_socket( nw_channel_h  ch)
{

  return  nw_create_socket_internal( &ch->socket);
}

/**************************************************************************
 *
 * Set "listen" state
 *
 * Parameters:
 *
 * IN nw_channel_h  ch      - pointer to a channel descriptor.
 * IN const char *  nw_dev  - Listener channel name
 * 
 * Description:
 * 
 * Sets the IO channel to the listener state.
 *
 * Return:
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
MCO_RET nw_listen( nw_channel_h  ch,  const char *  nw_dev)
{
  MCO_RET     ret;
  int         res;

    if( (ret=nw_create_socket(ch)) != MCO_S_OK) {
      return ret;
    }

    ch->sin.sin_family      = AF_INET; 
    ch->sin.sin_addr.s_addr = htonl(INADDR_ANY);
    ch->sin.sin_port	    = htons((int2)atoi(nw_dev));

    if ( bind((SOCKET)(ch->socket), (LPSOCKADDR)&ch->sin, sizeof(ch->sin)) != 0 )
    {
      nw_close(ch);
      return (MCO_RET)(MCO_E_NW_FATAL);
    }

    if ( listen((SOCKET)(ch->socket), SOMAXCONN) != 0 )
    {
      nw_close(ch);
      return (MCO_RET)(MCO_E_NW_BUSY);
    }

    NBIO(ch->socket, &res);
    if (res < 0 ) {
      nw_close(ch);
      return (MCO_RET)(MCO_E_NW_FATAL);
    }

    ch->status |= NWST_LISTEN;
    return MCO_S_OK;
}


/***************************************************************************
 *
 * Accept the connection
 *
 * Parameters:
 *
 * IN  nw_channel_h chan      - pointer to the listener channel.
 * IN OUT nw_channel_h  ioch  - pointer to the IO channel waiting for connection.
 * IN  unsigned long timeout) - wait-for-connect timeout.
 * 
 * Description:
 * 
 * Waits for the connection of a remote host to the IO channel.
 *
 * Return:
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
MCO_RET nw_accept( nw_channel_h chan,
                   nw_channel_h ioch,
                   timer_unit timeout
                 )
{
  int         csin_len;
  TIMEVAL     tv, *ptv = NULL;
  fd_set      readfds;
  int         rcs;
  conn_msg_t  msg;
  timer_unit      time = timeout;
  timer_unit      t1, t2;
   

  init_channel( ioch );

  if(timeout != MCO_TM_INFINITE) {
    t1 = t2 = mco_system_get_current_time();
    ptv = &tv;
  }

  for(;;) {

    FD_ZERO(&readfds);
    FD_SET((SOCKET)chan->socket, &readfds);

    if(timeout != MCO_TM_INFINITE) {
      if( ((t2=mco_system_get_current_time())-t1) > time) {
        return (MCO_RET)(MCO_E_NW_TIMEOUT);
      }
      time -= t2-t1;
      t1 = t2;
      
      tv.tv_sec = time/1000;
      tv.tv_usec= (time%1000)*1000;
    }
    else {
      tv.tv_sec = -1;
      tv.tv_usec= -1;
    }

    rcs = select (1+(int)chan->socket, &readfds, 0, 0, ptv);

    if( rcs < 0 ) // error
    {
      if(errno == EINTR) continue;
      nw_close(chan);
      return (MCO_RET)(MCO_E_NW_ACCEPT);
    }

    if( !rcs ) // timeout
    {
      nw_close(chan);
      return (MCO_RET)(MCO_E_NW_TIMEOUT);
    }

    if ( !(FD_ISSET((SOCKET)chan->socket, &readfds) ) )
    {
ErrRet:
      nw_close(chan);
      return (MCO_RET)(MCO_E_NW_ACCEPT);
    }

    if(timeout != MCO_TM_INFINITE) {
      if( (((t1=mco_system_get_current_time())-t1)) > time) {
          return (MCO_RET)(MCO_E_NW_TIMEOUT);

      }
      time -= t1;
    }

    csin_len = sizeof(chan->sin);
    ioch->sin.sin_addr.s_addr = htonl(INADDR_ANY);

    if (
         (ioch->socket =
           (SOCKET)accept(
			(SOCKET)chan->socket,
                        (LPSOCKADDR)&ioch->sin,
                        &csin_len)
         ) <0 
       )
    {
      goto ErrRet;
    }
    
    rcs = 1;

/* copy IP address & port to listener channel in order to give
   the listener thread opportunity to get them from the 
   listener channel descriptor
 */
    chan->sin.sin_addr.s_addr = ioch->sin.sin_addr.s_addr;
    chan->sin.sin_port	      = ioch->sin.sin_port;

    setsockopt (ioch->socket, IPPROTO_TCP, TCP_NODELAY,
      (const char*)&rcs, sizeof(int) );
    
    NBIO(ioch->socket, &rcs);
    if (rcs < 0 ) {
      goto ErrRet1;
    }

    ioch->status |= (NWST_CONNECTED|NWST_INITIALIZED);
    msg.index = ioch->index=allocate_index(); // channel index

    if(nw_send (ioch, (PCHAR)&msg, sizeof(conn_msg_t), time) != MCO_S_OK) {
ErrRet1:
       nw_close(ioch);
       return (MCO_RET)(MCO_E_NW_ACCEPT);
    }
    return MCO_S_OK;
  }  
}

/***************************************************************************
 *
 * Connect to another host
 *
 * Parameters:
 *
 * IN nw_channel_h   chan       - pointer to a channel descriptor
 * IN const char connect_string - interconnect dependent connection-string
 * IN  unsigned long timeout);
 * 
 * Description:
 * 
 * Connects IO channel to the remote host by it's name.
 *
 * Return:
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
MCO_RET nw_connect( nw_channel_h chan,
                    const char * connect_string,
                    timer_unit timeout
                  )
{
  int           rcs;
  const char *cc = connect_string;
  char name[128];
  int           len;
  conn_msg_t    msg;
  MCO_RET       rc;
  TIMEVAL       tv, *ptv = NULL;
  fd_set        readfds;
  timer_unit    time = timeout;
  timer_unit    t1, t2;

/*
  struct addrinfo *pres;
*/
  struct hostent *he;

  if(timeout != MCO_TM_INFINITE) {
    t1 = t2 = mco_system_get_current_time();
    ptv = &tv;
  }
  chan->status = 0;

/* Translate interconnect-dependent connect-string
 * Format: "xxx.xxx.xxx.xxx:nnnn" or "inet-name:port" */
/* Find ':' in connect string */

    for (;;) {
      if ( *cc == '\0' ) return (MCO_RET)(MCO_E_NW_INVADDR);
      if ( *cc == ':' ) break;
      cc++;
    }

    cc++;
    len = cc-connect_string;

    if ((int)(cc-connect_string) >= (int)sizeof(name) ) return (MCO_RET)(MCO_E_NW_NOMEM);
    strncpy(name, connect_string, (int)(cc-connect_string-1));
    name[(int)(cc-connect_string-1)] = 0;
/*
 * Translate IP address
 */
    if((chan->sin.sin_addr.s_addr = hostGetByName (name))==ERROR) {
       chan->sin.sin_addr.s_addr = inet_addr(name);
       return (MCO_RET)(MCO_E_NW_INVADDR);
    }

    chan->sin.sin_family = AF_INET; 
    chan->sin.sin_port = htons((uint2)atoi(cc));

    if( (rc=nw_create_socket(chan)) != MCO_S_OK) {
      return rc;
    }
    
    NBIO(chan->socket, &rcs);
    if (rcs < 0 ) {
ErrRet: nw_close(chan);
        return (MCO_RET)(MCO_E_NW_CONNECT);
    }

    for(;;) {  

      FD_ZERO(&readfds);
      FD_SET((SOCKET)chan->socket, &readfds);

      if(timeout != MCO_TM_INFINITE) {

        if( ((t2=mco_system_get_current_time())-t1) > time) {
          return (MCO_RET)(MCO_E_NW_TIMEOUT);