master.c

extremeDB s sample code,useful for you

/*****************************************************************
 *                                                               *
 *  master.c                                                     *
 *                                                               *
 * This file is a part of the eXtremeDB-HA Application Framework *
 *   It demonstrates McObject High Availabitity support and      *  
 * explains how to work with it.                                 *
 *                                                               * 
 *****************************************************************
 *  Copyright (c) 2001-2006 McObject LLC                         * 
 *  All Rights Reserved                                          *
 *****************************************************************/
/*
 * ++
 * 
 * PROJECT:   Ha Framework
 *
 * SUBSYSTEM: HA support
 *
 * MODULE:    master.c
 *
 * ABSTRACT:  The framework for HA support
 *
 *
 * VERSION:   1.0
 *
 * HISTORY:
 *            1.0- 1 AD     01-Mar-2003 Created it was
 *                 2 AG     19-Mar-2003 Ported to Linux
 *                 3 SS     08-Jul-2003 Reconstructed
 *                 4 AK     14-Dec-2004 mco_.quad from struct to scalar
 *
 * --
 */

/*
 * YF and performance test. This code writes many objects
 * of various types in HA mode
 *
 * Schema for this sample is in monitors.mco file.
 *
 * In order to change the maximum database size change DBSIZE
 * (currently 1M)
 */
#include "app.h"
#include "monitorDB.h"
#include "mcoHA.h"
#include "sensor.h"
#include "watchdog.h"

DECLARE_PROCESS_MASKS(set);
THREAD_PROC_DECLARE(ListenToReplicas);  // extern declaration

extern  int       conn_flag;
extern  int       main_master;
extern  mco_db_h  ReplDb;
extern  long       stop_flag;
extern  ha_h      haInstances[MAX_HA_INSTANCES];
extern  int       NumberOfDb;       // current number of db instances
extern  int       wdt_flag;         // the flag for shared commit watchdog
extern  int       p2_commit;        // 2-phase commit flag
extern  long      synch_flag;
extern  int       async_quantum;
extern mco_HA_params_t MasterParams;
extern  timer_unit  t_begin, t_start, t_stop, t_end;

uint4             NumBytes    = 0;
MCO_RET           rc;

/**************************************************
 * master's commit cycle
 **************************************************/

void master( int flag, ha_h ha )
{
  int             i;
  MCO_RET         rc;
  mco_trans_h	    t;
  unsigned long   counter;
  int2            nr;
  mco_uquad       auto_oid;
  timer_unit      time = 0, time1,time3 = 0;//, time2
  mco_db_h        db = ha->db;

  /* build so many databases as it is set in the option in command line */
  Printf( "\n\n\nMaster: Building the database, please wait\n" );

  /* fill out the database with some initial data */
  if(flag) {
      Printf("\nMaster:  *** Database %d ***", ha->id);

      Printf("\nMaster: populating initial data for Sensor1...");

      s1_init(db, S1_RECORDS);
      wdt_flag = 0;
      Printf("done, still available %dK\n",free_mem(  db,0 ));
  }
    mco_get_current_auto_oid(db, &auto_oid);
#ifdef	MCO_CFG_QUAD_STRUCT
  Printf("Master: Starting with auto oid: %d:%d\n",(int) auto_oid.hi,(int) auto_oid.lo);
#else  /* MCO_CFG_QUAD_STRUCT */
  Printf("Master: Starting with auto oid: %lld\n", auto_oid);
#endif /* MCO_CFG_QUAD_STRUCT */

  /* main loop - switch on sensors */

  Printf( "\nMaster: processing data, press Enter to terminate... \n");

  time1 = mco_system_get_current_time();

  for (i=0, counter=1;; counter++, i++ )
  {

//    if(!(i%5))
//      Sleep(5);

#ifdef _VXWORKS
    taskDelay(0);	// for VxWorks only, in order to optimize scheduling
#endif

#ifdef PRINT_REPORTS
    if(!i) time  = mco_system_get_current_time();
#endif

    rc = mco_trans_start( db, MCO_READ_WRITE, MCO_TRANS_FOREGROUND, &ha->t );
    if ( rc )
    { 
      Printf( "\nMaster: error starting write transaction %d\n", rc );
      exit( 1 );
    }
    t = ha->t;
    s1_new_record(t, 1000000+counter);
    if (stop_flag)
    {
      rc = mco_trans_rollback(t);
      break;
    }

#ifdef MCO_CFG_HA_2PHASE_COMMIT    
    if(p2_commit) {
    /* 2-phase commit is on */
      if(MCO_S_OK != (rc = mco_trans_commit_phase1(t)))
      { 
        Printf( "\nMaster: error commiting transaction %d\n", rc );
        EXIT(-1);
      }
      if(MCO_S_OK != (rc = mco_trans_commit_phase2(t)))
      { 
        Printf( "\nMaster: error commiting transaction %d\n", rc );
        EXIT(-1);
      }
    }
    else {
#endif //MCO_CFG_HA_2PHASE_COMMIT    
      if(MCO_S_OK != (rc = mco_trans_commit(t)))
      { 
        Printf( "\nMaster: error commiting transaction %d\n", rc );
        EXIT(-1);
      }
#ifdef MCO_CFG_HA_2PHASE_COMMIT    
    }
#endif //MCO_CFG_HA_2PHASE_COMMIT    

    wdt_flag = 0;    // clear commit watchdog flag
    if (stop_flag)  break;

#ifdef PRINT_REPORTS
    if(i==REPORT_COUNTER) {

      if (main_master) {
//        mco_HA_keep_alive( db, 10000);   // send signal "KEEP_ALIVE" to replicas
                                        // in order to give them sign that master is "alive"
      }
      if( (time = mco_system_get_current_time() - time) ==0) time++;
      time3 = (time3 + (1000*i/time))/2;
      nr = mco_HA_get_number_of_replicas( db );
      Printf(
        "Master: %ld ms for %d transactions (%d TPS);"
        " number of replicas: (%d) %d \n",
        time/NumberOfDb, i, (int)time3*NumberOfDb,
        ha->id+1, nr
      ); //(int)1000*i/time
      i = -1;

      if(MasterParams.mode_flags & MCO_HAMODE_ASYNCH) {
        Sleep(async_quantum);
      }

 #ifdef _VXWORKS
  	  Sleep(50);
 #endif
    }
#endif    
  }
  Printf("\n*** Master complete ***\n");
  Printf ("\n\n");
  t_stop = mco_system_get_current_time();

  if(ha->db != NULL)
  {
    mco_uquad auto_oid;
    mco_get_current_auto_oid(ha->db, &auto_oid);
#ifdef	MCO_CFG_QUAD_STRUCT
    Printf("Master: Communication over %s channel, last autoid = %d:%d\n",
      CHANNEL_NAME, (int)auto_oid.hi, (int)auto_oid.lo);
#else  /* MCO_CFG_QUAD_STRUCT */
    Printf("Master: Communication over %s channel, last autoid = %lld\n",
      CHANNEL_NAME, auto_oid );
#endif /* MCO_CFG_QUAD_STRUCT */

    MasterReport();

  }

  synch_flag--;

}

void  MasterReport()
{
  int i;
  FILE *        f;
  mco_db_h      db;
  mco_trans_h	  t;    // transaction handle
  MCO_RET       rc;
  char          s[200];

  /* output data for comparison with replica */
    Printf("\nMaster: initializing verification:\n");

    sprintf(s, "%smonitor.out",LOG_PATH);
  
    if((f = fopen(s, "w"))==0) {
      Printf ("  Master:  error opening file monitor.out\n");
    }
    else {
      Printf ("  Master: file %s is opened\n",s);

      for( i=0; i < NumberOfDb; i++) {
        ha_h ha;

        ha = haInstances[i];
        db = ha->db;

        if ( (rc=mco_trans_start( db, MCO_READ_ONLY, MCO_TRANS_FOREGROUND, &ha->t )) )
        {
          Printf( "\nMaster: error starting read transaction %d\n", rc );
          continue;
        }
        t = ha->t;
        fprintf(f,"database instance %d\n", i);
        Printf("writing data...\n");
        s1_output(t, f);
        Printf ("  Master: Sensor1 data is written\n");

#ifdef MCO_CFG_HA_2PHASE_COMMIT    
        if(p2_commit) {
          if(MCO_S_OK != (rc = mco_trans_commit_phase1(t)))
          { 
            Printf( "\nMaster: error commiting transaction %d\n", rc );
            EXIT(-1);
          }
          if(MCO_S_OK != (rc = mco_trans_commit_phase2(t)))
          { 
            Printf( "\nMaster: error commiting transaction %d\n", rc );
            EXIT(-1);
          }

        }
        else {
#endif //MCO_CFG_HA_2PHASE_COMMIT    
          if(MCO_S_OK != (rc = mco_trans_commit(t)))
          { 
            Printf( "\nMaster: error commiting transaction %d\n", rc );
            EXIT(-1);
          }
#ifdef MCO_CFG_HA_2PHASE_COMMIT    
        }
#endif //MCO_CFG_HA_2PHASE_COMMIT    
      }
      fclose( f );

    }

}