pipesw32.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:    pipesw32.C
 *
 * ABSTRACT:  Low level OS dependent IPC.
 *            Channel-over-pipes, platform dependent functions
 *
 *
 * VERSION:   1.0
 *
 * HISTORY:
 *            1.0- 1 SS     19-Feb-2004 Created it was
 *
 * --
 */

#ifdef _WIN32

#define INCLUDE_SOCKETS
#include "interface.h"

#ifdef CFG_PIPE_CHANNEL

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

extern uint4   NumBytes;

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

/***************************************************************************
 * 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 void init_channel( nw_channel_h chan )
{
    chan->index  = -1;
    chan->status = 0;
    chan->hndl   = INVALID_HANDLE_VALUE;
    chan->event  = INVALID_HANDLE_VALUE;
}


/***************************************************************************/
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 (see above).
 */
{
  if(chan->hndl != INVALID_HANDLE_VALUE) CloseHandle(chan->hndl);
  if(chan->event!= INVALID_HANDLE_VALUE) CloseHandle(chan->event);
   deallocate_index(chan->index);
   init_channel(chan);
   return MCO_S_OK;
}


/***************************************************************************/
MCO_RET nw_init(nw_channel_h chan)
/*
 * IN 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  = -2;
  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.
 */
  return MCO_S_OK;
}

/**************************************************************************/
MCO_RET nw_listen(nw_channel_h  chan,  const char *  nw_dev)
/*
 * IN nw_channel_h  chan    - pointer to a channel descriptor.
 * IN const char *  nw_dev  - listener channel name
 * 
 * Description:
 * Sets the IO channel to the listener state.
 * Format of the channel name is "\\.\pipe\##name##"
 * where "name" is user defined name
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
{
  int         len;
  char        lsnr_name[NW_FULL_NAMELENGTH];
#ifndef _WIN32
  uint4       mode = 1;
  #ifdef _SOLARIS
    int       flags;
  #endif
#endif  

    chan->index = -2;
    if( (len = strlen(nw_dev)) > NW_MAX_NAMELENGTH-1 ) len = NW_MAX_NAMELENGTH-1;
    memcpy( chan->name, nw_dev, len);
    chan->name[len] = 0;
    nw_close(chan);

    sprintf(lsnr_name, "%s", chan->name);

    /* create named server's listener channel */
    if ( (chan->hndl = CreateNamedPipe(lsnr_name,
                           FILE_FLAG_OVERLAPPED|PIPE_ACCESS_DUPLEX|0x80000, /*FILE_FLAG_FIRST_PIPE_INSTANCE, */
                           PIPE_TYPE_BYTE|PIPE_READMODE_BYTE|PIPE_WAIT,
                           1, 1024, 1024, 0, 0)) == INVALID_HANDLE_VALUE ) {
      nw_close(chan);
      return (MCO_RET)(MCO_E_NW_ERROR);
    }
    chan->event = CreateEvent( NULL, 0, 0, NULL);
    chan->status |= NWST_LISTEN;
    return MCO_S_OK;
}

/***************************************************************************/
MCO_RET nw_accept(nw_channel_h chan,
                  nw_channel_h  ioch,
                  timer_unit timeout)
/*
 * IN  nw_channel_h chan      - pointer to the listener channel.
 * IN OUT nw_channel_h  ioch  - pointer to 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.
 * 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).
 */
{
  uint4         ret;
  short         recvlen;
  char          ch_name[NW_FULL_NAMELENGTH];
  nw_pipename_t pipename;
  MCO_RET rc;

    /* get current channel index */

    init_channel( ioch );
    if( (ioch->index = allocate_index()) == -1) return (MCO_RET)(MCO_E_NW_FATAL);
    nw_close(ioch);
    ioch->index = allocate_index();
    memcpy(ioch->name, chan->name, NW_MAX_NAMELENGTH);

    ConnectNamedPipe(chan->hndl, 0);

    pipename.index = ioch->index;
    sprintf(ch_name, "%s_%d", chan->name, pipename.index);

#ifdef  NW_DEBUG_OUTPUT
    Printf("Pipe = %s\n",ch_name);
#endif

    /* create named IO channel */
    if ( (ioch->hndl = CreateNamedPipe(ch_name,
                         FILE_FLAG_OVERLAPPED|PIPE_ACCESS_DUPLEX|0x80000, /*FILE_FLAG_FIRST_PIPE_INSTANCE,*/
                         PIPE_TYPE_BYTE|PIPE_READMODE_BYTE|PIPE_WAIT,
                         1, 1024, 1024, 0, 0)) == INVALID_HANDLE_VALUE ) {
      goto ErrRet;
    }

    if ( !(WriteFile(chan->hndl, &pipename, sizeof(pipename), (PDWORD)&ret, 0)) ||
        (ret != sizeof(pipename)))
    {
ErrRet:
      nw_close(chan);
      nw_close(ioch);
      return (MCO_RET)(MCO_E_NW_ACCEPT);
    }
    if ( (rc = nw_recv(chan, (PCHAR)&pipename, sizeof(pipename), &recvlen, timeout)) != MCO_S_OK )   {
      nw_close(chan);
      nw_close(ioch);
      return rc;
    }
  
    DisconnectNamedPipe(chan->hndl);
    nw_close(chan);
    ioch->event = CreateEvent( NULL, 0, 0, NULL);
    ioch->status = (NWST_CONNECTED|NWST_INITIALIZED);
    return MCO_S_OK;
}

/***************************************************************************/
MCO_RET nw_connect(nw_channel_h chan,
                   const char * connect_string,
                   timer_unit timeout)
/*
 * IN nw_channel_h   chan       - pointer to channel descriptor
 * IN const char connect_string - interconnect dependent connection-string
 * IN  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).
 */
{
  uint4         ret;
  short         recvlen;
  char          lsnr_name[NW_FULL_NAMELENGTH];
  nw_pipename_t pipename;
  nw_channel_t  lch;
  MCO_RET rc;

   init_channel(&lch);
   init_channel(chan);

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

/*
 * Translate interconnect-dependent connect-string
 * Format: "\\.\pipe\pipename" or "\\Host\pipe\pipename"
 */
    sprintf(lsnr_name, "%s", lch.name);
    if ( (lch.hndl = CreateFile(lsnr_name,
                                  GENERIC_READ|GENERIC_WRITE,
                                  FILE_SHARE_READ|FILE_SHARE_WRITE,
                                  0,
                                  OPEN_EXISTING,
                                  FILE_FLAG_OVERLAPPED|FILE_ATTRIBUTE_NORMAL,
                                  0)) == INVALID_HANDLE_VALUE ) {
ErrRet0:
      perror("errno = ");
      nw_close(&lch);
      return (MCO_RET)(MCO_E_NW_CONNECT);
    }

    lch.event = CreateEvent( NULL, 0, 0, NULL);
    if ( (rc = nw_recv(&lch, (PCHAR)&pipename, sizeof(pipename), &recvlen, timeout)) != MCO_S_OK ) {
      goto ErrRet0;
    }

    /* get current channel index */
    chan->index = pipename.index;
    nw_close(chan);
    chan->index = pipename.index;

    sprintf(lsnr_name, "%s_%d", lch.name, pipename.index);
#ifdef  NW_DEBUG_OUTPUT
    Printf("Pipe = %s\n",lsnr_name);
#endif

    if ( (chan->hndl = CreateFile(lsnr_name,
                                  GENERIC_READ|GENERIC_WRITE,
                                  FILE_SHARE_READ|FILE_SHARE_WRITE,
                                  0,
                                  OPEN_EXISTING,
                                  FILE_ATTRIBUTE_NORMAL,
                                  0)) == INVALID_HANDLE_VALUE )
      goto ErrRet1;

    if ( !WriteFile(lch.hndl, &pipename, sizeof(pipename), (PDWORD)&ret, 0) ) {
ErrRet1:
      nw_close(chan);
      goto ErrRet0;
    }

    ConnectNamedPipe(chan->hndl, 0);
    chan->event = CreateEvent( NULL, 0, 0, NULL);
    lch.index = -1;
    chan->index = pipename.index;
    nw_close(&lch);
    return MCO_S_OK;
}

/***************************************************************************/
MCO_RET nw_send (nw_channel_h chan,
                 char* buffer,
                 int2 buflen,
                 timer_unit timeout)
/*
 * IN nw_channel_h  chan    - pointer to channel descriptor.
 * IN  const char*  buffer  - buffer to send.
 * IN int2 buflen           - buffer length,is cut to 32767 bytes.
 * IN  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).
 */
{
  uint4 ret;

  if ( !WriteFile(chan->hndl, buffer, buflen, (PDWORD)&ret, 0) || ret != (uint4)buflen ) {
     nw_close(chan);
     perror("write:");
     return (MCO_RET)(MCO_E_NW_SENDERR);
  }
  NumBytes += (uint4)buflen;
  return MCO_S_OK;
}

/***************************************************************************/
MCO_RET nw_recv (nw_channel_h chan,
                 char* buffer,
                 int2 buflen,
                 int2* recvlen,
                 timer_unit timeout)
/* IN nw_channel_h chan     - pointer to channel descriptor.
 * OUT char* buffer         - buffer to receive.
 * IN int2 buflen           - buffer length limit,  is cut to 32767 bytes.
 * OUT int2* recvlen,       - actual received length, is cut to 32767 bytes.
 * IN 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).
 */
 {
  uint4           len = (uint4) buflen;
  long            nbytes;
  uint4           res;
  OVERLAPPED      ovlp;

    while (len) {
      ovlp.hEvent = chan->event;
      ovlp.Offset = 0;
      ovlp.OffsetHigh = 0;
      if ( !ReadFile(chan->hndl, buffer, len, (PULONG)&nbytes, &ovlp) ) {
        int err;
        if((err=GetLastError()) != ERROR_IO_PENDING) {
          nw_close(chan);
          return (MCO_RET)(MCO_E_NW_RECVERR);
        }
      }
      res = WaitForMultipleObjects(1, &chan->event, 0, timeout );

      if( res != WAIT_OBJECT_0 ) {
        int err;
        err=GetLastError();
        nw_close(chan);
        return (MCO_RET)(MCO_E_NW_TIMEOUT);
      }
      res=GetOverlappedResult(chan->hndl, &ovlp, (PDWORD)&nbytes, 1);

      len -= nbytes;
      buffer += nbytes;
    }
    *recvlen = (short) buflen;
    NumBytes += (uint4)buflen;
    return MCO_S_OK;
}

/*
 * Not implemented. The stubs for the compliance with other platforms
 */

/***************************************************************************
 *
 * Accept the cancel point connection
 *
 * 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).
 */
MCO_RET nw_accept_cancel_point(
                   nw_channel_h chan,
                   char * port,
                   timer_unit timeout
                   )
{
    return MCO_S_OK;
}


/***************************************************************************
 *
 * Connect to cancel point
 *
 * 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.
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
MCO_RET nw_connect_cancel_point( nw_channel_h chan,
                                 const char * connect_string,  
                                 timer_unit timeout
                               )
{
    return MCO_S_OK;
}

/************************************************************************
 *
 * Cancels the communication channel
 *
 * Parameters:
 *
 * IN nw_channel_h ch      - pointer to a channel descriptor.
 *
 */
void  nw_cancel(nw_channel_h ch)
{
}


#endif// CFG_PIPE_CHANNEL

#endif /* _WIN32 */