haqmsg.c

extremeDB s sample code,useful for you

 * 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
 */
{
#ifdef  NW_DEBUG_OUTPUT
    Printf("\nHA error handler called, errcode = %d\n", HAerror->errcode);
#endif
    if(HAerror->IOchannel != 0) {   // if (replica was deleted) {}
      nw_close((nw_channel_h)HAerror->IOchannel); // close communication channel
//      free(HAerror->IOchannel);  // free the memory allocated by the descriptor
    }
}

/***************************************************************************/
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_db_h db       - database descriptor pointer.
 * const char*       - devname.
 * const mco_connection_param_t* params - connection & transaction timeouts.
 * const char* replica_name - replica name (optional).
 * 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.
 *
 * In channel-over-QNX-messaging module this function is called only once and never returns
 * to caller program. After it's been called it becomes message server.
 *
 * Returns:  hever
 */
{
  MCO_RET		ret;
//  nw_channel_t		_ch;        // copy of IO channel in the stack
  nw_channel_h		ch;         // pointer to the new IO channel
  pthread_t		DispatchThread;
  ha_h            ha = (ha_h)arg;
	
  baseCh =  &ha->baseChan;
    if( ! ha->isMainMaster )         // if master is passive then forbid creation
      commit_initialized = 1; // of shared commit thread
/*
 * If the transport protocol isn't initialized
 * then initialize it
 */
    if ( !(ha->baseChan.status&NWST_INITIALIZED) )  {
//    	nw_init(&ha->baseChan);
      ha->baseChan.db     = db;               // database handle. Is used by mco_nw_close()
      memcpy(&master,&db,sizeof(master_t)); // save arguments for creation of replicas

#ifdef NW_DEBUG_OUTPUT
      Printf( "\n  Initializing master communication channel's context, please wait...\n" );
#endif
      pthread_create((pthread_t*)&DispatchThread, NULL, 
                     (void*(*)(void*))Dispatcher, NULL);
/*
 * 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.
 */

      Sleep(2000);

#ifdef  NW_DEBUG_OUTPUT
      Printf("done\n");
#endif
    }
    ha->baseChan.db     = db;               // database handle. Is used by mco_nw_close()
    ha->baseChan.status |= NWST_IS_MASTER;  // ID_MASTER flag used by function mco_nw_close()
                                      // to check whether the transport layer still
                                      // belongs to master 
/*
 *  Waiting for connection request
 */
#ifdef  NW_DEBUG_OUTPUT
    Printf( "\n  Listener: Waiting for the connection request...\n" );
#endif
    if((ch=FindFreeChannel())==0) {
        ret  = (MCO_RET)(MCO_E_NW_BUSY);
        goto ErrRet;
    }

    if ( (ret = nw_accept(&ha->baseChan, ch, timeout)) != MCO_S_OK )
    {
#ifdef  NW_DEBUG_OUTPUT
ErrRet:
      if(ret != MCO_E_NW_TIMEOUT) 
        Printf("Error accepting connection %d\n", ret);
      else
        Printf("\nConnection timeout\n");
#endif
      return ret;
    }
#ifdef  NW_DEBUG_OUTPUT
    Printf("connection accepted\n");
#endif
/*
 * Creating and initializing channel descriptor
 */
    ch->mco_channel.fsend = (mco_xstream_write)&mco_nw_send; // set pointer to virtual HA send-receive methods
    ch->mco_channel.frecv = (mco_xstream_read)&mco_nw_recv; // to it's internal descriptor

/*
 *  Connection request is just arrived, attaching to replica
 */
#ifdef  NW_DEBUG_OUTPUT
    Printf( "  Attaching to replica, please wait...\n" );
#endif
    init_time  = mco_system_get_current_time();
    conn_flag = 1;
    ret = mco_HA_attach_replica(db, &ch->mco_channel, params, replica_name, ErrorHandler);
    init_time  = mco_system_get_current_time() - init_time;
    conn_flag = 0;
#ifdef  NW_DEBUG_OUTPUT
    if(ret)
      Printf("Error attaching to replica %d\n", ret);
#endif

    return ret;
}

/***************************************************************************/
MCO_RET mco_nw_attach_master(
                             mco_db_h*    db,
                             const char*  conn_string,
                             const mco_connection_param_t* params,
                             MCO_E_HA_REPLICA_STOP_REASON* stop_reason,
                             const char*  db_name,
                             mco_dictionary_h dict,
                             void*        mem_ptr,
                             uint4        total_size,
                             timer_unit   timeout,
                             void *       arg)
/*
 * mco_db_h*	db              - pointer to database handle
 * const char*conn_string     - interconnect dependent connection-string.
 * const mco_connection_param_t* params - connection & transaction timeouts.
 * MCO_E_HA_REPLICA_STOP_REASON* stop_reason - the reason why the replica is stopped.
 * const char*	db_name       - database name.
 * mco_dictionary_h dict      - pointer to database dictionary
 * void*	mem_ptr             - pointer to the memory allocated for database copy
 * uint4 total_size           - size of allocated memory
 * unsigned long timeout      - wait-for-connection timeout
 * IN OUT void * arg          - user defined argument. In this context it implements
 *                              special mode flag:
 *   flag != 0 - replica loads master's database and then becomes master
 *
 * Description:
 * This function includes connection algotrithm dependent on the type of a communication
 * channel. It attaches the replica to the connected master.
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
{
	MCO_RET   ret;
  char      name[NW_MAX_NAMELENGTH];
   ha_h            ha = (ha_h)arg;

  baseCh =  &ha->baseChan;

    *stop_reason = MCO_HA_REPLICA_HANDSHAKE_FAILED;
    commit_initialized++;
/*
 * If transport layer  isn't initialized  
 * then initialize it
 */
    if ( !(ha->baseChan.status & NWST_INITIALIZED) )  {
#ifdef NW_DEBUG_OUTPUT
      Printf( "\n  Initializing master communication channel's context, please wait...\n" );
#endif
      if( (ret=nw_init(&ha->baseChan))!=MCO_S_OK) {
#ifdef  NW_DEBUG_OUTPUT
        Printf("Error initializing: %d\n",ret);
#endif
        return ret;
      }
#ifdef  NW_DEBUG_OUTPUT
      Printf("done\n");
#endif
    }
/*
 * Connecting to master
 */
#ifdef  NW_DEBUG_OUTPUT
    Printf("Connecting...\n");
#endif
    if ( (ret=nw_connect(&ha->baseChan, conn_string,  timeout)) != MCO_S_OK )
    {
#ifdef  NW_DEBUG_OUTPUT
      Printf("Error connecting: %d\n", ret);
#endif
      return ret;
    }
#ifdef  NW_DEBUG_OUTPUT
      Printf("done\n");
#endif
/*
 * Initializing channel descriptor
 */
    ha->baseChan.mco_channel.fsend = (mco_xstream_write)&mco_nw_send;
    ha->baseChan.mco_channel.frecv = (mco_xstream_read)&mco_nw_recv;
/*
 *  Connection request is accepted, attaching to master
 */

#ifdef  NW_DEBUG_OUTPUT
    Printf( "  Attaching to master...\n" );
#endif

/*
   This sample code compiles the unique replica database name from parameter
   "db_name" and unique replica index obtained from master.
   please remove the line below if you want to replicate existing database
   and use "db_name" instead of "name"
  */
//***
    sprintf(name, "%s%d", db_name, (int)ha->baseChan.index);
    strcpy((char*)db_name,name);
    mco_db_kill(name);
//***

    Printf("DB name = %s\n",name);
    
    ret = mco_HA_attach_master(
                db,                   // pointer to database handle
                (mco_channel_h)&ha->baseChan, // pointer to the communication channel
                params,               // connection & transaction timeouts
                stop_reason,          // stop reason
                name,                 // database name
                dict,                 // pointer to database dictionary
                mem_ptr,              // pointer to the memory allocated for database copy
                total_size,           // size of allocated memory
                (uint2)((ha->replicaMode &MCO_HAFLAG_FORCE_MASTER) |
                MCO_HAFLAG_REPLICA_NOTIFICATION)
                );
    nw_close(&ha->baseChan);
    reallocate_index(ha->baseChan.pMsg->hd.id);
    commit_initialized = 0;
    return ret;
}
/***************************************************************************/
MCO_RET mco_nw_close (void *arg)
/*
 * IN OUT void * arg    - user defined argument. Not used in this context
 *
 * Description:
 * If called by the replica the function closes the communication channel.
 * if called by the master then  disconnects all replicas
 * connected to the master and closes all master/replica communication channels
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
{
  nw_channel_h  ch;
  short         nr;
  MCO_RET       rc;
   ha_h            ha = (ha_h)arg;
   
  baseCh =  &ha->baseChan;

    if ( ha->baseChan.status & NWST_IS_MASTER )  {
      if(ha->isMainMaster != 0) { // if (main master)
        for(nr=mco_HA_get_number_of_replicas((mco_db_h)ha->baseChan.db); nr; nr-- ) {
          Printf("Master: detaching replica %d ...", nr-1);
          ch = (nw_channel_h)mco_HA_get_io_channel((mco_db_h)ha->baseChan.db, nr-1);
          if ((rc = mco_HA_detach_replica((mco_db_h)ha->baseChan.db, (mco_channel_h)ch))
              != MCO_S_OK) {
	          Printf( "failed (%d)\n", rc );
          }
          /* no need to close channel, it's closed by ErrorHandler */
        }
      }
    }
    nw_close(&ha->baseChan);
    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 //_QNX