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
 *
 * AUTHORS:   Alexey Deriy    (AD)
 *            Andrei Gorine   (AG)
 *            Sergei Semanin  (SS)
 *
 * 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
 *
 * --
 */

/*
 * 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 "perf2.h"
#include "mco.h"
#include "watchdog.h"

#define   msec          mco_system_get_current_time

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       nRecords;//200000
long               synch_flag;
extern  timer_unit t_begin, t_start, t_stop, t_end;

uint4             NumBytes    = 0;
#ifdef CFG_SHARED_COMMIT
  static mco_HA_params_t mode = {MCO_MULTIPROCESS_COMMIT, 3};
#else
  static mco_HA_params_t mode = {MCO_YES, 3};
#endif

/*
  Master notification routine
*/

void master_notifying(
         uint2  notification_code,  /* notification code*/
         uint4  param1,             /* reserved for special cases */
         void * param2,             /* current transaction handle */
         void * pcontext            /* user-defined context of the notification*/
         )
{
}

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

void master( int flag, ha_h ha)
{
  int             i, curr_id = 0;
  MCO_RET         rc;
  mco_trans_h	    t;
  unsigned long   counter;
	timer_unit      start_time;
	long 		        n = 0;
	Record 		      rec;
	mco_cursor_t    c;
  mco_db_h        db;
  uint4           key;

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

  db  = ha->db;
  key = 1999;
  mco_HA_set_mode(db, &mode);  // set MASTER mode

    /* main loop - switch on sensors */

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

  for (counter=1;; counter++ )
  {

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

  /* insert Records, don't create the tree index yet
   */
  printf("(db %d)Insert", ha->id+1);

  Sleep(10);

    start_time = msec();
    n = 0;
    for (i=0; i < nRecords; i++) {
      key = ((uint4)db) + (2712818u*(i+1)) % 1000007u;
      rc = mco_trans_start( db, MCO_READ_WRITE, MCO_TRANS_FOREGROUND, &t );
      if( MCO_S_OK != rc )
	      exit(1);
      rc = Record_new(t, &rec);
      Record_key_put(&rec, key);

      rc = mco_trans_commit(t);

      if (!rc ) n++;
      if ((i%nRecords/10)==0)
		    printf (".");
      wdt_flag = 0;    // clear commit watchdog flag
    }

    {
      int t1;
      if((t1 = (int)(msec() - start_time ))==0) t1++;
	    printf(" %d objects: %d milliseconds, \n(%d microsecs/object, %d transactions/sec)\n",
		    n, t1, t1 * 1000/n,
        1000*n/t1);
    }

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

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

	    /* hash search */
    Sleep(10);
    rc = mco_trans_start( db, MCO_READ_ONLY, MCO_TRANS_FOREGROUND, &t );
    if( MCO_S_OK != rc ) {
        goto  exit;
    };

    printf("(db %d)Hash search ", ha->id+1);
    start_time = msec();
    for (i=0; i < nRecords; i++) {
      key = ((uint4)db) + (2712818u*(i+1)) % 1000007u;
      rc = Record_hkey_find(t, key, &rec);
      if (i%(nRecords/10)==0)
	      printf (".");
    }
    mco_trans_commit(t);

    if (main_master) {
      mco_HA_keep_alive( db, 10000);   // send signal "KEEP_ALIVE" to replicas
    }

    {
      int t1;
      if((t1 = (int)(msec() - start_time ))==0) t1++;
	printf(" %d objects: %d milliseconds,\n (%d microsecs/object, %d transactions/sec)\n",
		nRecords, t1, t1 * 1000/nRecords, 
        1000*nRecords/t1);
    }

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

	    /* tree search */

    Sleep(10);

    printf("(db %d)Tree search ", ha->id+1);
    start_time = msec();

    rc = mco_trans_start( db, MCO_READ_ONLY, MCO_TRANS_FOREGROUND, &t );
    Record_tkey_index_cursor(t, &c );

    for (i=0; i < nRecords; i++) {
      key = ((uint4)db) + (2712818u*(i+1)) % 1000007u;
      Record_tkey_search( t, &c, MCO_EQ, key );

      if ((i%nRecords/10)==0)
	      printf (".");

    }
    mco_trans_commit(t);

    if (main_master) {
      mco_HA_keep_alive( db, 10000);   // send signal "KEEP_ALIVE" to replicas
    }

    {
      int t1;
      if((t1 = (int)(msec() - start_time ))==0) t1++;
	printf(" %d objects: %d milliseconds,\n (%d microsecs/object, %d transactions/sec)\n",
		nRecords, t1, t1 * 1000/nRecords, 
        1000*nRecords/t1);
    }

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

    /* cursor movements */

    Sleep(10);

    printf("(db %d)Sequential ", ha->id+1);
    start_time=msec();
    rc = mco_trans_start( db, MCO_READ_ONLY, MCO_TRANS_FOREGROUND, &t );
    Record_tkey_index_cursor(t, &c );

    for (n = 0,rc = mco_cursor_first(t, &c);    rc == MCO_S_OK;  rc = mco_cursor_next(t, &c)) {
      if (n%(nRecords/10)==0)
	      printf (".");
      n++;
    }

    mco_trans_commit(t);

    if (main_master) {
      mco_HA_keep_alive( db, 10000);   // send signal "KEEP_ALIVE" to replicas
    }

    {
      int t1;
      if((t1 = (int)(msec() - start_time ))==0) t1++;
	printf(" %d objects: %d milliseconds,\n (%d microsecs/object, %d transactions/sec)\n",
		nRecords, t1, t1 * 1000/nRecords, 
        1000*nRecords/t1);
    }

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

    /* Search using hash index and remove the object ones found
     */
    printf("(db %d)Search/delete ", ha->id+1);

    start_time = msec();

    for (n=0,i=0; i < nRecords; i++) {
      key = ((uint4)db) + (2712818u*(i+1)) % 1000007u;
      rc = mco_trans_start( db, MCO_READ_WRITE, MCO_TRANS_FOREGROUND, &t );
      if( MCO_S_OK != rc ) exit(1);

      rc = Record_hkey_find(t, key, &rec);
      rc = Record_delete(&rec);
      rc = mco_trans_commit(t);

      if (!rc) n++;
      if ((n%nRecords/10)==0)
		    printf (".");
      wdt_flag = 0;    // clear commit watchdog flag
    }

    {
      int t1;
      if((t1 = (int)(msec() - start_time ))==0) t1++;
	printf(" %d objects: %d milliseconds,\n (%d microsecs/object, %d transactions/sec)\n\n",
		nRecords, t1, t1 * 1000/nRecords, 
        1000*nRecords/t1);
//      if(t1 >= 5000 ) nRecords /= 10;
    }

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

  Printf("\n*** Master complete ***\n");
  
  t_stop = mco_system_get_current_time();

  Printf ("\n\n");
  {
    mco_uquad auto_oid;
    mco_get_current_auto_oid(haInstances[0]->db, &auto_oid);
#ifdef MCO_CFG_QUAD_STRUCT    
    Printf("Communication over %s channel, last autoid = %d:%d\n",
      CHANNEL_NAME, (int)auto_oid.hi, (int)auto_oid.lo);
#else
    Printf("Communication over %s channel, last autoid = %lld\n",
      CHANNEL_NAME, auto_oid );
#endif
  }

exit:
  synch_flag--;
}

void    MasterReport()
{
}