replica.cpp

extremeDB s sample code,useful for you

/*****************************************************************
 *                                                               *
 * This file is a part of the eXtremeDB-HA Application Framework *
 * Copyright (c) 2001-2006 McObject LLC                          * 
 * All Rights Reserved                                           *
 *                                                               *    
 *****************************************************************
 * ++
 * 
 * PROJECT:   eXtremeDB(tm) (c) 2003 McObject LLC
 *
 * SUBSYSTEM: HA framework
 *
 * MODULE:    replica.cpp
 *
 * ABSTRACT:  C++ implementation of replica mode
 *
 *
 * VERSION:   1.0
 *
 * HISTORY:
 *            1.0- 1 SS     18-Feb-2004 Created it was
 *
 * --
 *****************************************************************/
#include  "mcoha.hpp"

extern void make_strings( void );

static char *	master_search_list[] ={
  nw_ConnectionString1,
	nw_ConnectionString2,
	nw_ConnectionString3,
#if(SECOND_INTERFACE)
  nw_ConnectionString4,
  nw_ConnectionString5
#endif  
};

static void replica_notifying(
               uint2  notification_code,  /* notification code*/
               uint4  param1,             /* reserved for special cases */
               void * param2,             /* reserved for special cases */
               void * pcontext            /* user-defined cnotification context */
               )
{

  switch(notification_code) {
    default:
      Printf("\r\n** Notification ** Replica's been notified code = %d, param1 = %ld, param2 = %08lx\r\n",
         notification_code,  /* notification code*/
         param1,             /* reserved for special cases */
         param2              /* reserved for special cases */
        );
      break;
    case MCO_REPL_NOTIFY_CONNECTED:         /* "connected" notification */
      Printf("\r\n** Notification ** Replica's been connected\r\n"
        );
      break;
    case MCO_REPL_NOTIFY_CONNECT_FAILED:    /* "connect failed" notification */
      Printf("\r\n** Notification ** Connection failed error = %d\r\n",
         param1
        );
    case MCO_REPL_NOTIFY_DB_EQUAL:          /* "no need to load DB" notification */
      Printf("\r\n** Notification ** Master & Replica databases are equal\r\n"
        );
      break;
    case MCO_REPL_NOTIFY_DB_LOAD_BEGIN:     /* "begin loading DB" notification */
      Printf("\r\n** Notification ** Replica's begun loading\r\n"
        );
      break;
    case MCO_REPL_NOTIFY_DB_LOAD_FAILED:    /* "loading failed" notification, "param1" of
                                            notification callback contains MCO_RET code */
      Printf("\r\n** Notification ** Database loading failed error = %d\r\n",
         param1
        );
      break;

    case MCO_REPL_NOTIFY_DB_LOAD_OK:        /* "succesful loading" notification, "param1" of
                                            notification callback contains
                                            MCO_E_HA_REPLICA_STOP_REASON code */
      Printf("\r\n** Notification ** Database's been loaded OK\r\n"
        );

#ifdef CFG_SYNCHRON_EVENTS
      {
        MCO_RET rc;
  	    if ((rc = register_events(Replha->db)) != MCO_S_OK) {
		      printf("Error registering events, %d\n", rc);
	      }
      }
#endif //CFG_SYNCHRON_EVENTS

      break;

    case MCO_REPL_NOTIFY_COMMIT_FAILED:     /* "commit failed" notification, "param1" of
                                            notification callback contains MCO_RET code */
      Printf("\r\n** Notification ** Commit failed error = %d\r\n",
         param1
        );
      break;

    case MCO_REPL_NOTIFY_REPLICA_STOPPED:    /* "stopped" notification, */
      {
        char *reason;
        switch (param1)
        {
          default:
            reason = "Replica stopped by the keysroke";
            break;
          case  MCO_HA_REPLICA_HANDSHAKE_FAILED:
            reason = "MCO_HA_REPLICA_HANDSHAKE_FAILED";
            break;
          case MCO_HA_REPLICA_CONNECTION_ABORTED:
            reason = "MCO_HA_REPLICA_CONNECTION_ABORTED";
            break;
          case  MCO_HA_REPLICA_MASTER_REQUESTED_DISCONNECT:
            reason = "MCO_HA_REPLICA_MASTER_REQUESTED_DISCONNECT";
            break;
          case MCO_HA_REPLICA_STOPPED_BY_LOCAL_REQUEST:
            reason = "MCO_HA_REPLICA_STOPPED_BY_LOCAL_REQUEST";
            break;
          case MCO_HA_REPLICA_BECOMES_MASTER:
            reason = "MCO_HA_REPLICA_BECOMES_MASTER";
            break;
        }
        Printf ("\n** Notification ** Replica stopped, reason: %d (%s)\n",
          param1,
          reason);

      }
      break;
  }
}

/******************************************************
 * Replica Mode
 *
 * Creates the only database instance. Runs replica mode
 ******************************************************/

int   App::ReplicaMode()
{
  int           i;
  MCO_RET				rc;
  FILE					* f;
  char          s[200];
  MCO_E_HA_REPLICA_STOP_REASON stop_reason;
  Database *    Db;

/* 
 * create the only instance of HA for replication. Database isn't open yet after the
 * ccreation of this object.
 */
    Db=DbInstances[0]= new Database(CurrentDbInstance);

    InitDBinstance( Db, CurrentDbInstance );

    for(i=0;!i;) {
      isReplica = 1;        // replica mode on
      Db->db = 0;

      /* attach to master */
      stop_reason = Replica();

      switch(stop_reason) {
        default:
          goto ExitReplica;
        case MCO_HA_REPLICA_CONNECTION_ABORTED:
        case MCO_HA_REPLICA_BECOMES_MASTER:
          if ( replica_mode >= 0 )   // analize -s switch
          {
            Printf("\n*** Replica switching to master ***\n");
          /*
           * This timeout is used in order to give the kernel
           * opportunity to detach master's socket from the faulty process
           * before the running replica on the same machine uses the socket.
           */
            Sleep(1000);
            t_begin = GetCurrentTime(); // save test begining time
            Db->isMainMaster = main_master = 1;
            t_start = GetCurrentTime();
      	    wdt_flag = 0;         // turn on watchdog
            isReplica = 0;        // replica mode off
            Db->InitThreads();    // init master's threads
            Master( 0 );          // flag = 0 - database is already initialized
            return 0;
          }
        case MCO_HA_REPLICA_HANDSHAKE_FAILED:
          Printf("Clearing memory...\n");
          if(db != 0) {
            DbDisconnect();
            DbClose(Db->dbName);
          }
          Printf("Reconnecting to the new master...\n");
          Sleep(RECONNECTION_DELAY+(GetCurrentTime()&0x3ff) );  // randomize access
          break;
        case MCO_HA_REPLICA_STOPPED_BY_LOCAL_REQUEST:
        case MCO_HA_REPLICA_MASTER_REQUESTED_DISCONNECT:
          i++;    // break cycle
          break;
      }
    }

    Printf("\n*** Replica complete ***\n");
    Printf( "\nGenerating reports, please wait..." );
    for ( i=1; ; i++ )
    {
      sprintf(s, "monrepl%d.out", i);
      if((f = fopen(s, "r"))) {
        fclose(f);
      }
      else break;
      
    }
    f = fopen(s, "w");
    Printf ("\n  File %s is opened\n", s);
    rc = Db->StartTransaction( MCO_READ_ONLY, MCO_TRANS_FOREGROUND);
    if ( rc )
    {
		
      Printf( "\nerror starting read transaction %d\n", rc );
      exit( 1 );
    }
    Db->s1.output(f);
    Printf ("  Sensor1 data is written\n");

    Db->TransCommit();

ExitReplica:

    isReplica = 0;        // replica mode off
    return  0;
}

/*
 * Runs replica mode. Attaches to master, makes the copy of master's database 
 * and then performs internal replica loop.
 */

MCO_E_HA_REPLICA_STOP_REASON App::Replica()
{
  char            * reason = 0;
  Database*       Db = DbInstances[0];
  MCO_RET         rc;
  int             rcc;
  int             i;
  int StopReason   = -1;
  mco_runtime_info_t    run_info;
  char            mastername[HOST_MAX_NAMELENGTH];

  mco_get_runtime_info( &run_info);

  /* set connection & HA timeouts */
  Db->conpar.commit_timeout   = TM_INFINITE;        // timeout for commit
  Db->conpar.initial_timeout  = TM_ATTACH_REPLICA;  // initial sync timeout
  Db->conpar.detach_timeout   = TM_REPLICA_COMMIT;  // send detach request,  timeout for replica's commit
  Db->conpar.wait_data_timeout= TM_WAIT_FOR_DATA;   // amount of time replica waits for the next commit
  Db->conpar.notifying_callback = replica_notifying;

  /* try to connect to one of the masters from the list */
  for(i=0;i<sizeof(master_search_list)/sizeof(char*);i++) {
    if(stop_flag) {
      rc = (MCO_RET)-1;
      StopReason = -1;
      break;
    }

    /* add HA instance index to master logical name */
    strcpy(mastername, master_search_list[i]);
    sprintf(&mastername[strlen(mastername)-2],"%02d", Db->id*10);

    Printf ("Open communication channel %s, Attaching to master (blocking call) .... \n",
					  mastername);

    /* reset database name */
    sprintf(Db->dbName,"monitorDB%d",Db->id);
    Db->db = 0;
    
    /* try to attach to current master */
    rc =	Db->AttachMaster(
			   mastername,
			   (MCO_E_HA_REPLICA_STOP_REASON*)&StopReason,
			   Db->dbName,
			   monitorDB_get_dictionary(),
                           Db->start_mem,
                           DBSIZE,
                           TM_CONNECTION_TIMEOUT
                           );
		db = Db->db;

    if ( rc )
    {
      Printf( "\nError attaching master %d\n", rc );

      /* is the error legal? */
      switch(rcc = rc) {

        default:
           if((rcc >= MCO_ERR_NW) && (rcc <= MCO_ERR_NW+99))   continue;
           break;
        case MCO_E_NW_FATAL:
        case MCO_E_NW_NOTSUPP:
        case MCO_E_NW_BUSY:
        case MCO_E_NW_NOMEM:
            EXIT(-1);
      }
    }
    break;
  }

  /* print the reason by wich replica is disconnected from master */
  switch (StopReason)
  {
    default:
      reason = "Replica stopped by the keysroke";
      break;
    case  MCO_HA_REPLICA_HANDSHAKE_FAILED:
      reason = "MCO_HA_REPLICA_HANDSHAKE_FAILED";
      break;
    case MCO_HA_REPLICA_CONNECTION_ABORTED:
      reason = "MCO_HA_REPLICA_CONNECTION_ABORTED";
      break;
    case  MCO_HA_REPLICA_MASTER_REQUESTED_DISCONNECT:
      reason = "MCO_HA_REPLICA_MASTER_REQUESTED_DISCONNECT";
      break;
    case MCO_HA_REPLICA_STOPPED_BY_LOCAL_REQUEST:
      reason = "MCO_HA_REPLICA_STOPPED_BY_LOCAL_REQUEST";
      break;
    case MCO_HA_REPLICA_BECOMES_MASTER:

      reason = "MCO_HA_REPLICA_BECOMES_MASTER";
      break;
  }
  Printf ("\nReplica stoped, reason: %d (%s)\n", StopReason, reason);
  return (MCO_E_HA_REPLICA_STOP_REASON)StopReason;
}