udpvx.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:    udpvx.c
 *
 * ABSTRACT:  Low level OS dependent IPC.
 *            Channel-over-UDP, 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_UDP_SOCKET_CHANNEL

#include "ioLib.h"

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

extern
MCO_RET _nw_send_message(
                nw_channel_h   chan,
                uint1          type,
                const void*    buffer,
                int2           buflen,
                timer_unit     timeout);
extern
MCO_RET _nw_recv_message(
                nw_channel_h   chan,
                uint1          type,
                const void*    buffer,
                int2           buflen,
                int2*          recvlen,
                timer_unit     timeout);

extern uint4        NumBytes;

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

udp_msg_buf_t msgbuf;

/****************************************************************************
 * 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    * 
 *                                                                          *
 *          Remark:                                                         *
 * Transfer protocols usually use frame descriptors (headers) for data      *
 * transfer control.                                                        *
 *   Each created UDP channel MUST have a framebuffer for                   *
 * data transfer.                                                           *
 *                                                                          *
 * The framebuffer has the following format:                                *
 *                                                                          *
 *                                                                          *
 *        -------------------- <- pointer to framebuffer                    *
 *        |   frame header   |                                              *
 *        |------------------| <- (pointer to framebuffer)+(header length)  *
 *        |   message body   |                                              *
 *        --------------------                                              *
 *                                                                          *
 *   Application MUST create it's own framebuffers and attach them to each  *
 * created channel with the function nw_attach_buffer()                     *
 *                                                                          *
 *  Mco HA uses the one and only internal static buffer for all master's    *
 * channels or for replica's channel.                                       *   
 ****************************************************************************/

/***************************************************************************/
void  nw_attach_buffer( nw_channel_h chan, PVOID buffer, uint2 bufsize)
/*
 *  attaches framebuffer to a channel
 */
{
    chan->pMsg = (udp_msg_buf_h)buffer;
    chan->bufsize = (uint2)(bufsize + sizeof(udp_msg_buf_t));
}

/******************************************************************************
 * function init_channel() - initializes the channel descriptor
 */
static MCO_RET init_channel(nw_channel_h ch)   /* pointer to channel descriptor */
{
    ch->recv_seqn = 0;
    ch->send_seqn = 0;
    ch->destaddr  = 0;
    ch->pMsg      = (udp_msg_buf_h)&msgbuf;
    ch->bufsize   = sizeof(udp_msg_buf_t);
    ch->currlen   = 0;
    ch->index     = -1;
    ch->status    = 0; 
    ch->socket        = INVALID_SOCKET_VALUE;
    ch->cancel_socket = INVALID_SOCKET_VALUE;
    return  MCO_S_OK;
}

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

    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;
    }
    deallocate_index(chan->index);
    init_channel(chan);
    return  rc;
}

/**************************************************************************
 *
 * Create nonblocked UDP 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).
 */
static MCO_RET create_socket( nw_channel_h chan, uint2 flag)
{
  int         optval=1, rcs;
  uint4       mode=1;

    /* open a DATAGRAM socket*/
    if ( (chan->socket = socket(PF_INET, SOCK_DGRAM, 0)) < 0 )  {
      nw_close(chan);
      return (MCO_RET)(MCO_E_NW_ERROR);
    }

    chan->sin.sin_family = AF_INET;               // Internet protocol stack
    chan->sin.sin_addr.s_addr  = htonl(chan->destaddr); // destination address
    chan->sin.sin_port	       = htons(chan->port);     // owner port number

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

    /* bind socket with the mentioned parameters */
    if ( bind(chan->socket, (LPSOCKADDR)&chan->sin, sizeof(chan->sin)) != 0 )
    {
      nw_close(chan);
      return (MCO_RET)(MCO_E_NW_ERROR);
    }
  
    NBIO(chan->socket, &rcs);
    if (rcs < 0 ) {
     nw_close(chan);
     return (MCO_RET)(MCO_E_NW_FATAL);
    }
    
    chan->status = (uint2)(NWST_INITIALIZED|flag); // base channel status = initialized 
    return MCO_S_OK;
}


/***************************************************************************
 * 
 * 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)
{
  atexit((void*)mco_nw_close);
/*
 * 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.
 */

  /* The WinSock DLL is acceptable. Proceed. */
   chan->status = NWST_INITIALIZED;       // base channel status = initialized 
   return MCO_S_OK;
}

/**************************************************************************
 *
 * 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  chan,  const char *  nw_dev)
{
  
    /* create UDP socket*/
    init_channel(chan);
    chan->port     =(int2)atoi(nw_dev);    // listener port number
    chan->destaddr = (uint4)INADDR_ANY;    // enable to listen to any host
    return create_socket(chan,(NWST_LISTEN|NWST_CONNECTED));
}


/***************************************************************************
 *
 * 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)
{
  int2        len;
  conn_msg_t  msg;
  MCO_RET     rc;
  timer_unit  time = timeout;
  timer_unit  t1, t2;

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

    init_channel(ioch);
    chan->recv_seqn  = 0;
    chan->send_seqn = 0;
    chan->pMsg  = (udp_msg_buf_h)&msg;
    chan->bufsize = sizeof(conn_msg_t);
    chan->destaddr = 0;

    if(
        (rc =
           _nw_recv_message(
                           chan,                // pointer to a channel object
                           FT_CONNECT,          // message type
                           &msg.body,           // receive buffer
                           sizeof(msg.body),    // receive buffer limit
                           &len,                // return actual length
                           time)                // connect timeout
        ) != MCO_S_OK
      ) 
    {
ErrRet0:
        nw_close(chan);
        return rc;
    }
/*
   at this point (SOCKADDR_IN)chan->sin contains IP address & port
   of the opponent in order to give the listener thread opportunity
   to get them from the listener channel descriptor
 */

    msg.body.index=(uint2)(ioch->index=allocate_index()); // channel index
    ioch->port = (uint2)(chan->port+((ioch->index+1)<<1));
    msg.body.port  = (uint2)(ioch->port);                  // IO channel port number
    ioch->destaddr = (uint4)INADDR_ANY;
    ioch->destport = (uint2)(ioch->port+1);

    /* create UDP socket*/
    if((rc=create_socket(ioch,NWST_CONNECTED)) != MCO_S_OK)                // create socket
    {
ErrRet:
      nw_close(ioch);
      goto ErrRet0;
    }
    ioch->destaddr = chan->inaddr;

    if(timeout != MCO_TM_INFINITE) {
      if( ((t2=mco_system_get_current_time())-t1) > time) {
        rc = (MCO_RET)(MCO_E_NW_TIMEOUT);
        goto ErrRet;
      }
      time -= t2-t1;
    }

    if(
        (rc=
           _nw_send_message(
                          chan,
                          FT_ACCEPT,
                          &msg.body,
                          sizeof(msg.body),
                          time)
         ) != MCO_S_OK
      )
    {
      goto ErrRet;
    }
    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)
{
  short         len;
  const char*   cc = connect_string;
  char          name[128];
  nw_channel_t  lch;
  conn_msg_t    msg;
  MCO_RET       rc;
  timer_unit  time = timeout;
  timer_unit  t1, t2;
  struct hostent *he;

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

/* 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++;
    if ( (cc-connect_string) >= 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;

    init_channel(&lch);