tcpwce.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:    tcpwce.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

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;

#ifdef CFG_SHARED_COMMIT

extern  THREAD_ID synchcommit;
THREAD_PROC_DECLARE(SynchCommit);

#endif  //  CFG_SHARED_COMMIT

static void init_channel( nw_channel_h chan )
{
    chan->index  = -1;
    chan->status = 0;
    chan->socket = -1;
#ifdef _WIN32
    chan->event  = NULL;
#endif
}

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

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

    deallocate_index(chan->index);
    if(chan->event != NULL) {CloseHandle(chan->event);
 chan->event = NULL;}
    if(chan->socket != -1) {if(closesocket(chan->socket)) rc = (MCO_RET)(MCO_E_NW_ERROR);}
    init_channel(chan);
    return rc;
}

/***************************************************************************/
MCO_RET nw_init( nw_channel_h chan)
/*
 * 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.
 *
 * Returns MSO_S_OK if successful or error code.
 */
{
  uint2 wVersionRequested;
  WSADATA wsaData;
  int err;

  wVersionRequested = MAKEWORD( 1, 1 );

  err = WSAStartup( wVersionRequested, &wsaData );
  if ( err != 0 ) {
    /* Tell the user that we could not find a usable */
    /* WinSock DLL.                                  */
    return (MCO_RET)(MCO_E_NW_NOTSUPP);

  }

  if ( LOBYTE( wsaData.wVersion ) != 1 ||
       HIBYTE( wsaData.wVersion ) != 1 ) {
    /* Tell the user that we could not find a usable */
    /* WinSock DLL.                                  */
    WSACleanup( );
    return (MCO_RET)(MCO_E_NW_NOTSUPP); 
  }

  /* The WinSock DLL is acceptable. Proceed. */
  init_channel( chan );
  chan->status = NWST_INITIALIZED;       // base channel status = initialized 

/*
 * Creation of the thread that implements shared commit. It is necessary
 * if you need to run several masters sharing the same database in the shared memory segment.
 */
#ifdef CFG_SHARED_COMMIT

  if(!commit_initialized) {
    Sleep(1);
    createThread(SynchCommit, (PVOID)chan->db, &synchcommit);
    commit_initialized++;
  }
#endif   //CFG_SHARED_COMMIT

  return MCO_S_OK;
}

/**************************************************************************/
MCO_RET nw_create_socket( nw_channel_h  ch)
/*
 * IN nw_channel_h  ch      - pointer to a channel descriptor.
 * 
 * Description:


 * Creates an overlapped TCP socket.
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
{
  int optval = 1;

    if ( (ch->socket = socket(PF_INET, SOCK_STREAM, 0)) < 0 )  {
      return (MCO_RET)(MCO_E_NW_FATAL);
    }

    setsockopt (ch->socket, IPPROTO_TCP, TCP_NODELAY,
      (const char*)&optval, sizeof(int) );
      
    setsockopt (ch->socket, SOL_SOCKET, SO_REUSEADDR,
      (const char*)&optval, sizeof(int) );

    return MCO_S_OK;
}

/**************************************************************************/
MCO_RET nw_listen( nw_channel_h  ch,  const char *  nw_dev)
/*
 * 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.
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
{
  SOCKADDR_IN sin;
  uint4       mode = 1;
  MCO_RET     ret;

    ch->port = (int2)atoi(nw_dev);

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

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

    if ( bind((SOCKET)(ch->socket), (LPSOCKADDR)&sin, sizeof(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);
    }

    if (ioctlsocket(ch->socket, FIONBIO, (PDWORD)&mode) < 0 ) {
      nw_close(ch);
      return (MCO_RET)(MCO_E_NW_FATAL);
    }
    ch->status |= NWST_LISTEN;
    return MCO_S_OK;
}


/***************************************************************************/
MCO_RET nw_accept( nw_channel_h chan,   nw_channel_h ioch,
                   uint4 timeout)
/*
 * 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.
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
{
  SOCKADDR_IN csin;

  int         csin_len = sizeof(csin);
  uint4       mode = 1;
  TIMEVAL     tv;
  fd_set      readfds;
  int         rcs;
  conn_msg_t  msg;
  timer_unit      time  = (long)timeout;
  timer_unit      t1, t2 = mco_system_get_current_time();

    init_channel( ioch );

  for(;;) {
      t1 = t2;
      if( (time-=((t2=mco_system_get_current_time())-t1)) <= 0) {
        return (MCO_RET)(MCO_E_NW_TIMEOUT);

      }

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

    tv.tv_sec = time/1000;
    tv.tv_usec= (time%1000)*1000;

	  rcs = select (0,           &readfds, 0, 0, &tv);

    if( rcs < 0 ) // error
    {
      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 ((ioch->socket = (SOCKET)accept((
                                  SOCKET)chan->socket,
                                  (LPSOCKADDR)&csin,
                                  &csin_len)) <0 ) {
      goto ErrRet;
    }
    
    rcs = 1;
    setsockopt (ioch->socket, IPPROTO_TCP, TCP_NODELAY,
      (const char*)&rcs, sizeof(int) );
    
    ioch->event = CreateEvent( NULL, 0, 0, NULL);

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

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