haqmsg.c

extremeDB s sample code,useful for you

/*******************************************************************
 *                                                                 *
 *  haQMSG.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:    haQMSG.c
 *
 * ABSTRACT:  Low level OS dependent IPC. Named pipes
 *
 * AUTHORS:
 *
 * VERSION:   1.0
 *
 * HISTORY:
 *            1.0- 1 SS     10-Sep-2003 Created it was
 *
 * --
 */

#ifdef _QNX

#define INCLUDE_SOCKETS
#include "interface.h"

#ifdef CFG_QNXMSG_CHANNEL

#include <stdlib.h>
#include <stdio.h>
#include <string.h>

#include <unistd.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/stat.h>
#include <sys/ioctl.h>
#include <semaphore.h>
#include <fcntl.h>
#include <errno.h>
#include <malloc.h>

#ifdef __cplusplus
extern "C" {
#endif

#ifdef __cplusplus
}
#endif

extern MCO_RET  mco_nw_close(void *arg);
extern int2     allocate_index();
extern void     deallocate_index(int2 index);
extern void     reallocate_index(int2 index);

static void ErrorHandler( ha_error_h  HAerror);
static int mco_nw_send(  nw_channel_h  ch,	  const void *  buffer,
                       unsigned buflen,	  timer_unit timeout);
static int mco_nw_recv(  nw_channel_h ch,   void*  buffer,
                       unsigned buflen,   timer_unit timeout);
extern uint4          NumBytes;

int                   conn_flag = 0;
timer_unit            init_time = 0;

static  master_t      master;
static  nw_channel_t  channels[MAX_CHANNELS]; // list of IO channels;
static int          commit_initialized = 0;
static nw_channel_h  baseCh;


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


static  nw_channel_h FindFreeChannel()
{
  int i;

  for(i=0; i < MAX_CHANNELS; i++) {
    if(channels[i].status == 0) return &channels[i];
  }
  return  0;
}

static  nw_channel_h FindChannelById(uint2 Id)
{
  int i;

  for(i=0; i < MAX_CHANNELS; i++) {
    if(channels[i].index == Id) return &channels[i];
  }
  return  0;
}

static void init_channel(nw_channel_h chan)
{
  chan->fd     = (uint4)-1;
  chan->rcv_id = (uint4)-1;
  chan->snd_id = (uint4)-1;
  chan->status = 0;
  chan->index  = (uint2)-1;
  chan->currlen = 0;            // current buffer length
}

/***************************************************************************/
MCO_RET nw_close( nw_channel_h chan)
/*
 *   nw_channel_h ch)      - pointer to the channel descriptor.
 *
 * Description:
 * Closes the communication channel
 * 
 * Returns MSO_S_OK if successful or error code (see above).
 */
{
    msg_header_t	msg;
    if(chan->fd !=-1) {
      if(chan->status&NWST_IS_MASTER) {  ;//ConnectDetach(chan->fd);
      	if(chan->status & NWST_RECV_PENDED) {
          MsgReply( chan->rcv_id, MCO_E_NW_RECVERR, &msg, 0);
      	}
	if(chan->status & NWST_SEND_PENDED) {
          MsgReply( chan->snd_id, MCO_E_NW_SENDERR, &msg, 0);
        }
      }
      else close(chan->fd);
    }
    deallocate_index(chan->index);
    init_channel(chan);
    return MCO_S_OK;
}

/***************************************************************************/
MCO_RET nw_init(  nw_channel_h chan)
/*
 *   nw_channel_h chan   - pointer to a 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 (see above).
 */
{
  init_channel(chan);
  chan->index = (uint2)-2;
  chan->status = NWST_INITIALIZED;       // base channel status = initialized
  return MCO_S_OK;
}


/***************************************************************************/
MCO_RET nw_accept(  nw_channel_h  chan,
                    nw_channel_h  ioch,
                    timer_unit    timeout)
/*
 *    nw_channel_h chan      - pointer to the listener channel.
 *     nw_channel_h  ioch  - pointer to IO channel waiting for connection.
 *    unsigned long timeout) - wait-for-connect timeout.
 *
 * Description:
 * Waits for the connection of a remote host to the IO channel.
 * Format of the IO channel name is "\\.\pipe\##name##_##number##"
 * where the "name" is user defined name
 * The "number" is the channel index in ASCII *
 * Returns MSO_S_OK if successful or error code (see above).
 */
{
  long             ret;
  conn_msg_t       msg;
  
    init_channel(ioch);

    /* get current channel index */

    if( ((short)ioch->index = allocate_index()) == -1) return (MCO_RET)(MCO_E_NW_FATAL);
    memcpy(ioch->name, chan->name, NW_MAX_NAMELENGTH);
    ioch->fd	   = chan->fd;
    msg.hd.id      = chan->index = ioch->index;
    msg.hd.type    = FT_ACCEPT;
    msg.hd.magic   = MAGIC;
    msg.hd.length  = 0;

    chan->timeout = (long)TIME_INFINITE;
    if( (ret=MsgSend(chan->fd , &msg, sizeof(msg), &msg, sizeof(msg)))<0) {
      nw_close(chan);
      return (MCO_RET)(MCO_E_NW_ACCEPT);
    }

#ifdef  NW_DEBUG_OUTPUT
    Printf( "\n  Accepting the connection request\n" );
#endif
#ifdef  NW_DEBUG_OUTPUT
    Printf("current channel index = %d\n", ioch->index);
#endif

    ioch->pMsg = (msg_buf_h)malloc(sizeof(msg_buf_t));
    ioch->timeout = (long)TIME_INFINITE;
    ioch->status = (NWST_INITIALIZED|NWST_CONNECTED|NWST_IS_MASTER);

    return MCO_S_OK;
}

/***************************************************************************/
MCO_RET nw_connect(  nw_channel_h chan,
                     const char * connect_string,
                     timer_unit   timeout)
/*
 *   nw_channel_h   chan       - pointer to channel descriptor
 *   const char connect_string - interconnect dependent connection-string
 *    unsigned long timeout);
 *
 * Description:
 * Connects IO channel to a remote host by it's name.
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
{
  long             st;
  char             ch_name[NW_FULL_NAMELENGTH];
  nw_channel_t     lch;
  conn_msg_t       msg;

    init_channel(&lch);
    init_channel(chan);
    memcpy(lch.name, connect_string, NW_MAX_NAMELENGTH);
    memcpy(chan->name, connect_string, NW_MAX_NAMELENGTH);

/*
 *  open named listener channel
 *
 * Translate interconnect-dependent connect-string
 * Format: "name" or "/net/host/name/"
 */
#ifdef  NW_DEBUG_OUTPUT
    Printf("listener channel name = %s\n",lch.name);
#endif

    if((lch.fd = open((char*)lch.name, O_RDWR)) < 0 ) {
ErrRet:
      perror("Connect error:");
      nw_close(&lch);
      return (MCO_RET)(MCO_E_NW_CONNECT);
    }

    msg.hd.type    = FT_CONNECT;
    msg.hd.magic   = MAGIC;
    msg.hd.length  = (uint2)(sizeof(conn_msg_t)-sizeof(msg_header_t));

    if((st=MsgSend(lch.fd ,&msg,sizeof(msg),
                           &msg,sizeof(msg)))  != 0 ) {
      goto ErrRet;
    }
    reallocate_index(chan->index = msg.hd.id);
#ifdef  NW_DEBUG_OUTPUT
    Printf("channel index = %d\n",msg.hd.id);
#endif
    chan->fd = lch.fd;
    chan->pMsg = (msg_buf_h)malloc(sizeof(msg_buf_t));
    chan->status = (NWST_INITIALIZED|NWST_CONNECTED);
    return MCO_S_OK;
}

/***************************************************************************/
MCO_RET nw_send (  nw_channel_h chan,
                   char*        buffer,
                   int2         buflen,
		   timer_unit   timeout)
/*
 *   nw_channel_h  chan    - pointer to channel descriptor.
 *    const char*  buffer  - buffer to send.
 *   int2 buflen           - buffer length,is cut to 32767 bytes.
 *    unsigned long timeout - send timeout in milliseconds. It is media dependent
 *                            & application dependent. Application MUST set this
 *                            parameter considering the media rate & it's own needs.
 *
 * Description:
 * Sends message to communication channel
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
{
  msg_header_t  answer;
  int           ret;
  msg_buf_h msgbuf = chan->pMsg;

    chan->timeout = (long)timeout/2;
    
    msgbuf->hd.type    = FT_SEND;
    msgbuf->hd.magic   = MAGIC;
    msgbuf->hd.id      = chan->index;

    if((msgbuf->hd.length=buflen) != 0)
      memcpy(msgbuf->buffer,buffer,(size_t)buflen);
//    Printf("Send Message %d bytes\n",(int)buflen);

      if((ret = MsgSend(chan->fd ,msgbuf, sizeof(msg_header_t)+(int)buflen,
	                (PVOID)&answer,sizeof(answer)))  < 0 ) {
     ret = errno;
//     perror("\nwrite:");
     nw_close(chan);
     return (MCO_RET)(MCO_E_NW_SENDERR);
    }
    chan->timeout = (long)TIME_INFINITE;
    NumBytes += (uint4)buflen;
//    Printf("sent %d bytes\n",(int)buflen);
    return MCO_S_OK;
}

/***************************************************************************/
MCO_RET nw_recv (  nw_channel_h chan,
                   char*        buffer,
		   int2         buflen,
                   int2*        recvlen,
		   timer_unit   timeout)
/*   nw_channel_h chan     - pointer to channel descriptor.
 *   char* buffer         - buffer to receive.
 *   int2 buflen           - buffer length limit,  is cut to 32767 bytes.
 *   int2* recvlen,       - actual received length, is cut to 32767 bytes.
 *   unsigned long timeout - receive timeout in milliseconds. It is media dependent
 *                            & application dependet. Application MUST set this parameter
 *                            considering the media rate & it's own needs
 *
 * Description:
 * Receives message from communication cchannel
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
 {
  conn_msg_t  msg;
  int         ret;
  msg_buf_h   msgbuf = chan->pMsg;

    chan->timeout = (long)timeout/2;
    msg.hd.type    = FT_RECV;
    msg.hd.magic   = MAGIC;
    msg.hd.length  = 0;
    msg.hd.id      = chan->index;

    if(chan->currlen) {
      if(chan->currlen < buflen) {