hatcp.c

extremeDB s sample code,useful for you

/*******************************************************************
 *                                                                 *
 * hatcp.c                                                         *
 *                                                                 *
 *  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:   hatcp.c
 *
 * ABSTRACT:  Low level OS dependent IPC. Channel-over-TCP
 *
 *
 * VERSION:   1.0
 *
 * HISTORY:
 *            1.0- 1 SS  19-Feb-2004 Creared it was
 *
 * --
 */

#define INCLUDE_SOCKETS
#include "interface.h"

extern  int         cancel_flag;    // turns on "replica cancellation" feature

/* 
 * Procedures for common use
 */
static int          map_initialized = 0;
static uint4        index_map[NW_MAX_INDEXES/32+(NW_MAX_INDEXES%32 ? 1:0)];

#ifdef __cplusplus
  extern "C" { 
#endif

/*
 * allocate free index. It's necessary to avoid repeated port/pipe numbers
 */
int2 allocate_index()
{
  int2 i,j;
  uint4  l;

  if(!map_initialized) {
    j = NW_MAX_INDEXES/32+(NW_MAX_INDEXES%32 ? 1:0);
    for(i=0; i < j; i++) index_map[i] = 0; 
    map_initialized = 1;
  }

  for(i=0,j=0, l=1; i < NW_MAX_INDEXES; i++) {
    if ( !(index_map[j] & l) ) {
      index_map[j] |= l;
      return i;
    }
    if( !(l<<= 1) ) {
      l = 1; j++;
    }
  }
  return  (int2)-1;
}

void deallocate_index(int2 index)
{
  uint4 l;

  if(index <0) return;
  l = 1 << (index % 32);
  index_map[index/32] &= ~l;
}

void reallocate_index(int2 index)
{
  int2 i,j;
  uint4 l;

  if(!map_initialized) {
    j = NW_MAX_INDEXES/32+(NW_MAX_INDEXES%32 ? 1:0);
    for(i=0; i < j; i++) index_map[i] = 0; 
    map_initialized = 1;
  }

  if(index <0) return;
  l = 1 << (index % 32);
  index_map[index/32] |= l;
}

#ifdef __cplusplus
  }
#endif

#ifdef CFG_TCP_SOCKET_CHANNEL

#ifdef __cplusplus
  extern "C" { 
#endif

extern uint4        NumBytes;
extern int          conn_flag;
extern timer_unit   init_time;
extern int          commit_initialized;

#ifdef CFG_SHARED_COMMIT

extern  THREAD_ID synchcommit;

#endif  //  CFG_SHARED_COMMIT

#ifdef __cplusplus
  }
#endif


/***************************************************************************
 * Interface layer. Used by master - replica interconnections.             * 
 * The interface is simplified. All the internal protocol details are      *
 * hidden behind this layer from the application and MCOdb HA              * 
 ***************************************************************************/

/******************************************************************************
 * 
 * Send method of HA runtime.
 * 
 * IN nw_channel_h  ch    - pointer to a channel descriptor.
 * IN const void*  buffer - buffer of data to send.
 * IN unsigned  buflen    - number of bytes to send
 * IN timer_unit timeout  - wait-for-send-completion timeout.
 *
 * Description:
 * This send function is used implicitly by master-replica interconnection channel.
 * The pointers to this function MUST be set to HA internal descriptor mco_channel_t
 * in each channel descriptor.
 *
 * Returns the actual length of sent data or -1 if error.
 */
static int mco_nw_send(
                  nw_channel_h  ch,
                  const void *  buffer,
                  unsigned      buflen,
                  timer_unit    timeout)
{
    if ( nw_send(ch, (char*)buffer, (int2)buflen, timeout) != MCO_S_OK ) {
      return -1;
    }
    return buflen;
}

/******************************************************************************
 *
 *  Receive method of HA runtime.
 *
 * IN  nw_channel_h ch      - pointer to a channel descriptor.
 * OUT void* buffer         - buffer for data to receive.
 * IN unsigned buflen       - number of bytes to receive.
 * IN  timer_unit timeout   - wait-for-receive-completion timeout.
 *
 * Description:
 * This receive function is used implicitly by master-replica interconnection channel
 * mco_nw_recv MUST read exactly max_bytes !!!
 * The pointers to this function MUST be set to HA internal descriptor mco_channel_t
 * in each channel descriptor.

 *
 * Returns the actual length of sent data or -1 if error.
 */
static int mco_nw_recv(
                  nw_channel_h ch,
                  void*  buffer,
                  unsigned buflen,
                  timer_unit timeout)
{
  uint2      currlen;

    if (
        nw_recv(
                ch,
                (char*)buffer,
                (int2)buflen,
                (int2*)&currlen,
                timeout
                )
        != MCO_S_OK 
       ) {
      return -1;
    }
    return buflen;
}

/***************************************************************************
 *
 * HA error handler.
 *
 * IN ha_error_h  HAerror - pointer to ha_error_t structure described in mcoha.h.
 * The structure contains error code that caused the handler call & pointer to 
 * the descriptor of communication channel of the deleted replica. If the pointer not
 * equal to NULL then replica was deleted from master's list of replicas.
 *
 * Description:
 * HA error handler callback - closes replica channel & deallocates memory
 * used by it's descriptor
 */
static void  ErrorHandler( ha_error_h  HAerror)
{
  nw_channel_h ch = (nw_channel_h)HAerror->IOchannel;

#ifdef  NW_DEBUG_OUTPUT
    Printf("Master: HA error handler called, errcode = %d\n", HAerror->errcode);
#endif
    if( ch != 0 ) {                 // if (replica was deleted) {}

#ifdef  NW_DEBUG_OUTPUT
      Printf("Master: replica addr %s port %d\n",
        inet_ntoa(ch->sin.sin_addr),
        ntohs(ch->sin.sin_port));
#endif
      
      nw_close( ch );               // close the communication channel
      free( ch );                   // free the memory allocated by the descriptor 
    }
}

/***************************************************************************
 *
 * Attaches replica to the master.
 *
 * IN	mco_db_h db       - database descriptor pointer.
 * IN	const char*       - devname.
 * IN const mco_connection_param_t* params - connection & transaction timeouts.
 * IN	const char* replica_name - replica name (optional).
 * IN unsigned long timeout    - wait-for-connection timeout
 * IN OUT void * arg    - user defined argument. In this context
 * it is used as main/slave master flag in multimaster mode
 * (for shared memory only)
 *
 * Description:
 * This function includes connection algotrithm dependent on the type of a communication
 * channel. It attaches connected replica to HA subsystem.
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
MCO_RET mco_nw_attach_replica(
	      mco_db_h db,
              const char* devname,
              const mco_connection_param_t * params,
              const char* replica_name,
              timer_unit timeout,
              void * arg)
{
  MCO_RET	  ret;
  nw_channel_t    _ch;        // copy of IO channel in the stack
  nw_channel_h    ch;         // pointer to the new IO channel
  char            name[NW_MAX_NAMELENGTH];
  ha_h            ha = (ha_h)arg;

/*
 * If the transport protocol isn't initialized  
 * then initialize it
 */
      ha->baseChan.db = db;               // database handle. Is used by mco_nw_close()
      if ( !(ha->baseChan.status&NWST_INITIALIZED) )  {
#ifdef NW_DEBUG_OUTPUT
        Printf( "\n  Initializing master communication channel's context, please wait..." );
#endif
        if( (ret = nw_init(&ha->baseChan)) != MCO_S_OK) {
#ifdef NW_DEBUG_OUTPUT
          Printf("Error initializing");
#endif
          return ret;
        }
      }

/* For POSIX/Linux only. WIN32 uses events. */
#ifndef _WIN32
   ha->baseChan.cancel_socket = INVALID_SOCKET_VALUE;
#endif //_WIN32
  
/*
 * If the master's listener channel isn't initialized  

 * then open the listener channel
 */