main.c

eXtremeDB数据库在Linux平台下的使用源码。内含多个示例程序

static int get_random_val( void )
{
/*
  static int num = 3;
  num = (num *19 + 25) % 3037;
  return num;
  */
  return rand();
}

static MCO_RET insertTrn(mco_db_h db)
{
	MCO_RET	    rc;
	mco_trans_h t;
  Rec         rec;
  uint4       key;
  char        buf[100];
	CheckSums * chk = ( CheckSums *) CHECKSUM_ADDR;
  int         recsum;
  uint2       buflen;
  uint2       vl = 0, i;
  Svalue      sv;
  char        * str;
  uint2       len;

  // Insert 1 record

  key = get_random_val() + idx;

  rc = mco_trans_start( db, MCO_READ_WRITE, MCO_TRANS_FOREGROUND, &t );
  if( MCO_S_OK != rc )
  {
    printf("\nError starting trn\n");
    return rc;
  }
  
  rc = Rec_new(t, &rec);
  if(rc)
    goto Err;

  rc = Rec_key_put(&rec, key);
  if( rc )
    goto Err;

  vl = rand2(1, MAX_VECT_LEN);
  if( (rc = Rec_valuev_alloc(&rec, vl)) )
    goto Err;
  
  for ( i=0; i < vl; i++ )
  {

    if( (rc = Rec_valuev_put(&rec, i, &sv)) )
      goto Err;
    if( (rc = Svalue_tm_put(&sv, i + idx)) )
      goto Err;
    if( (rc = Svalue_u4_put(&sv, i + idx)) )
      goto Err;
    str = get_random_string();
    len = (uint2) strlen(str);
    rc = Svalue_bytes_put(&sv, str, len);
  }

  sprintf( buf, "str %d", key);
  buflen = (uint2) strlen(buf);
  Rec_str_put(&rec, buf, buflen );

//  recsum = record_checksum(key, buf, buflen);
  recsum = record_checksum_byhandle(&rec);

  chk->checksum_before_commit = chk->checksum_after_commit ^ recsum;   

  rc = mco_trans_commit(t);
  if( rc == MCO_S_OK )
    chk->checksum_after_commit = chk->checksum_before_commit;
  if(rc)
    i++;
  idx += 0x123456;
  return rc;
 Err:
  mco_trans_rollback(t);
  idx += 0x123456;
  return rc;
}


static MCO_RET deleteTrn(mco_db_h db)
{
	MCO_RET	    rc;
	mco_trans_h t;
  Rec         rec;
	CheckSums * chk = ( CheckSums *) CHECKSUM_ADDR;
  int         recsum;
  mco_cursor_t csr;

  // Delete first record

    rc = mco_trans_start( db, MCO_READ_WRITE, MCO_TRANS_FOREGROUND, &t );
    if( MCO_S_OK != rc )
    {
      printf("\nError starting trn\n");
      return rc;
    }

    Rec_tkey_index_cursor(t, &csr);
    rc = mco_cursor_first(t, &csr);
    if(rc != MCO_S_OK)
    {
      // no records
      goto Err;
    }
    rc = Rec_from_cursor(t, &csr, &rec);
    if( rc )
       goto Err;

    recsum = record_checksum_byhandle(&rec);
  
    rc = Rec_delete(&rec);
    if(rc)
      goto Err;
    chk->checksum_before_commit = chk->checksum_after_commit ^ recsum;   
    rc = mco_trans_commit(t);
    if( rc == MCO_S_OK )
      chk->checksum_after_commit = chk->checksum_before_commit;
    return rc;

Err:
    mco_trans_rollback(t);
    return rc;
}

/****************************************************
 *          "Check Resources" thread 
 ****************************************************/

THREAD_PROC_DEFINE(CheckResources, arg)
{
  mco_db_h db = 0;
  unsigned wdt_state = 0;

    THREAD_PROC_MODE();

    printf("check time %ld\n", CheckTime);
    Sleep(CheckTime);

    for(;;) {

      if(stopflag) {
l10:    stopflag--;
      	printf("CheckResources thread complete\n");
        return;
      }

      Sleep(CheckTime);
      if(stopflag) goto l10;

      if(!startflag) continue;

    	if(!mco_db_check_resources( dbName, &db, 40000, &wdt_state))
        printf("\n*** recovery was done ***\n");
    }

}


/****************************************************
 *          "Working" thread 
 ****************************************************/

THREAD_PROC_DEFINE(Working, par)
{
  int creating_db = 0; //1;
  MCO_RET rc;
  int cnt = 0;

  int ndeletes = 0, nfailed_deletes = 0,
    ninserts=0, nfailedinserts = 0;

  THREAD_PROC_MODE();

  startflag = 0;
  Sleep(100);

  printf("\n****** Thread started ******\n");

  startflag++;

  creating_db  = starting_create_new_db(&db);

  if( creating_db  )
      printf("\nCreating DB for the first time\n");
  else
      printf("\nRecovering DB after power off\n");

  Sleep(500);

  rc = connect_to_db(&db, creating_db);

  if(rc)
  {
    printf("\nDb connecting error\n");
    for(;;);
  }

  printf("\nStarting cycle of inserts and deletes\n");

  for(cnt=0; !stopflag ;cnt++) // just creating & deleting records, updating checksum, until interrupted
  {
    int n;

    n = nRecords(db);

    count++;

    if(n > (20 + cnt % 10) )
    {
      ndeletes++;
      rc = deleteTrn(db);
      if(rc)
        nfailed_deletes++;
    }
    else
    {
      ninserts++;
      rc = insertTrn(db);

      if(rc)
        nfailedinserts++;
    }

    if( cnt % LOG_PERIOD == 0 )
    {
	    CheckSums * chk = ( CheckSums *) CHECKSUM_ADDR;
      classStat(db);
      printf("\nCurrent Checksum = %d (%d)\n",
        chk->checksum_after_commit, chk->checksum_before_commit);

      printf("\n ndeletes=%d, nfailed_deletes=%d, ninserts=%d, nfailedinserts=%d\n",
        ndeletes, nfailed_deletes, ninserts, nfailedinserts);
      Sleep(100);
    }

  }
  printf("\nWorking thread complete\n");
  mco_db_disconnect(db);
  mco_db_close(dbName);
  stopflag--;

}


void PowerOff()
{
  int i, j;

    startflag = 0;

    THREAD_CANCEL(thWorkThread);

    Sleep(100);

    printf("\n**************** RESTART *******************\n");
    printf("\n**************** RESTART *******************\n");
    printf("\n**************** RESTART *******************\n");

    for(i = 0; i <100; i++) {
      if(stopflag) goto l10;
      j = count;	
      Sleep(100);
      printf("\n*** Wait stopping working thread ***\n");
      if( count == j) break;
      count = 0;
    }	
    printf("\n*** Restart working thread ***\n");

    THREAD_PROC_START(Working, 0, &thWorkThread);

l10:;
}

/* parsing command line */
int parse_cmd(int argc, char **argv )
{
  int   i;
  char  *p;
  int   flag  = 1; 

  if (argc < 2)
  {
    return 1;
  }

  for (i=1;i<argc;i++)

  {
    p=argv[i];
    if ( (p[0] == '/')||(p[0] == '-') )
    {
      if(p[1] == 'h') {
        help();
        return 0;
      }
      if(p[1] == 't') {
        flag++;
        if( (CheckTime=atoi((const char*) &p[2])) == 0) {
          CheckTime = 1000;
        }
      }
    }
  }
  return flag;
}

#ifndef WIN32_WCE
int main(  int argc, char **argv )
#else
int __cdecl main(  int argc, char **argv )
#endif
{
  int i;
  FILE * f = stdin;
  char s[256];

    pdb = &db;

    PROCESS_MASKS();         // for LINUX, QNX, SOLARIS

#if defined(_LINUX) || defined(_QNX) || defined(_SOLARIS) 
   sigdelset(&set,SIGALRM);
   sigprocmask(SIG_SETMASK,&set, 0);
#endif

    s[0] = 0;

    {
      mco_runtime_info_t info;

      mco_get_runtime_info( &info );

      if( !info.mco_recovery_supported ) {
        printf("\n\n Recovery support is not configured! \n");
        PROG_EXIT(0);
      }
    }
  
    _SH_();

    if (parse_cmd(argc,argv )==0) {
#ifdef _WIN32_WCE
      char str[128];
      printf("\npress ENTER to exit\n");
      gets(str);
#endif
      PROG_EXIT(0);
    }

    mco_runtime_start();

    mco_error_set_handler( &errhandler );

    mco_runtime_setoption(
	  MCO_RT_OPTION_ZMUTEX_NR1, // loop  waiting the resource
	  long_loop        	    // the loop value
       );

    make_strings( );

    if(CheckTime) {
      THREAD_PROC_START(CheckResources, 0, &thCheckRes); /* run "Check Resources" thread */
    }

    THREAD_PROC_START(Working, 0, &thWorkThread);

    Sleep(1000);

    for(;;) {
#ifdef _VXWORKS
      i = 0;
      while(i == 0) {
	      ioctl (f->_file, FIONREAD, (int)&i);
	      Sleep(100);
      }
#endif
      if((fgets(s,256,stdin)) == 0) {
        continue;
      }
      if (s[0] ==0 ) continue;
      i = atoi(s);
      if ( i == 343 ) EXIT(1);
      if (s[0] == 'e' ) {
        stopflag = 2;
	break;
      }

      PowerOff();
    }

    Printf("\n\nApplication interrupted by keystroke\n");
    Printf("Wait for the completion of working threads\n");

    for(i=0; i< 15; i++) {

    if( !stopflag ) break;

//      Printf("%d running threads lost\n", stopflag);
      Sleep(100);  /* wait for terminating of working threads */
    }

    mco_runtime_stop();

    printf("program termination\n");

    PROG_EXIT(0);

}