benchronja.cpp

来自「MySQL数据库开发源码 值得一看哦」· C++ 代码 · 共 1,202 行 · 第 1/3 页

/* Copyright (C) 2003 MySQL AB

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 2 of the License, or
   (at your option) any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */


/* ***************************************************
       NODEREC
       Perform benchmark of insert, update and delete transactions

       Arguments:
        -t Number of threads to start, default 1
        -o Number of loops per thread, default 100000
	

 * *************************************************** */

#include <ndb_global.h>

#include <NdbApi.hpp>
#include <NdbTest.hpp>
#include <NdbOut.hpp>
#include <NdbThread.h>
#include <NdbSleep.h>
#include <NdbMain.h>
#include <NdbTimer.hpp>
#include <NdbTick.h>
#include <random.h>

#define MAX_TIMERS 4 
#define MAXSTRLEN 16 
#define MAXATTR 64
#define MAXTABLES 64
#define MAXTHREADS 256
#define MAXATTRSIZE 8000
#define START_TIMER NdbTimer timer; timer.doStart();
#define STOP_TIMER timer.doStop();
#define START_TIMER_TOP NdbTimer timer_top; timer_top.doStart();
#define STOP_TIMER_TOP timer_top.doStop();

void* ThreadExec(void*);
struct ThreadNdb
{
  int NoOfOps;
  int ThreadNo;
  Ndb* NdbRef;
};

static NdbThread* threadLife[MAXTHREADS];
static unsigned int tNoOfThreads;
static unsigned int tNoOfOpsPerExecute;
static unsigned int tNoOfRecords;
static unsigned int tNoOfOperations;
static int ThreadReady[MAXTHREADS];
static int ThreadStart[MAXTHREADS];

NDB_COMMAND(benchronja, "benchronja", "benchronja", "benchronja", 65535){
  ndb_init();

  ThreadNdb		tabThread[MAXTHREADS];
  int			i = 0 ;
  int			cont = 0 ;
  Ndb*			pMyNdb = NULL ; //( "TEST_DB" );	
  int           tmp = 0 ;
  int			nTest = 0 ;
  char inp[100] ;

  tNoOfThreads = 1;			// Default Value
  tNoOfOpsPerExecute = 1;	// Default Value
  tNoOfOperations = 100000;	// Default Value
  tNoOfRecords = 500 ;		// Default Value <epaulsa: changed from original 500,000 to match 'initronja's' default  
  i = 1;
  while (argc > 1)
  {
    if (strcmp(argv[i], "-t") == 0){
      tNoOfThreads = atoi(argv[i+1]);
      if ((tNoOfThreads < 1) || (tNoOfThreads > MAXTHREADS)) goto error_input;
    }else if (strcmp(argv[i], "-o") == 0){
      tNoOfOperations = atoi(argv[i+1]);
      if (tNoOfOperations < 1) goto error_input;
    }else if (strcmp(argv[i], "-r") == 0){
      tNoOfRecords = atoi(argv[i+1]);
      if ((tNoOfRecords < 1) || (tNoOfRecords > 1000000000)) goto error_input;
	}else if (strcmp(argv[i], "-p") == 0){
		nTest = atoi(argv[i+1]) ;
		if (0 > nTest || 18 < nTest) goto error_input ;
    }else if (strcmp(argv[i], "-c") == 0){
      tNoOfOpsPerExecute = atoi(argv[i+1]);
      if ((tNoOfOpsPerExecute < 1) || (tNoOfOpsPerExecute > 1024)) goto error_input;
    }else{
      goto error_input;
    }
    argc -= 2;
    i = i + 2;
  }

  ndbout << "Initialisation started. " << endl;
  pMyNdb = new Ndb("TEST_DB") ;
  pMyNdb->init();
  ndbout << "Initialisation completed. " << endl;

  ndbout << endl << "Execute Ronja Benchmark" << endl;
  ndbout << "  NdbAPI node with id = " << pMyNdb->getNodeId() << endl;
  ndbout << "  " << tNoOfThreads << " thread(s) " << endl;
  ndbout << "  " << tNoOfOperations << " transaction(s) per thread and round " << endl;

  if (pMyNdb->waitUntilReady(120) != 0) {
    ndbout << "Benchmark failed - NDB is not ready" << endl;
	delete pMyNdb ;
    return NDBT_ProgramExit(NDBT_FAILED);
  }//if

  NdbThread_SetConcurrencyLevel(2 + tNoOfThreads);

  for (i = 0; i < tNoOfThreads ; i++) {
    ThreadReady[i] = 0;
    ThreadStart[i] = 0;
  }//for

  for (i = 0; i < tNoOfThreads ; i++) {
    tabThread[i].ThreadNo = i;
    tabThread[i].NdbRef = NULL;
    tabThread[i].NoOfOps = tNoOfOperations;
    threadLife[i] = NdbThread_Create(ThreadExec,
                                       (void**)&tabThread[i],
                                       32768,
                                       "RonjaThread",
                                       NDB_THREAD_PRIO_LOW);
  }//for
  
  cont = 1;
  while (cont) {
	NdbSleep_MilliSleep(10);
    cont = 0;
    for (i = 0; i < tNoOfThreads ; i++)
      if (!ThreadReady[i]) cont = 1;
  }//while

  ndbout << "All threads started" << endl;
  
  if(!nTest){

	  for (;;){

		  inp[0] = 0;
		  ndbout << endl << "What to do next:" << endl;
		  ndbout << "1 \t=> Perform lookups in short table" << endl;
		  ndbout << "2 \t=> Perform lookups in long table" << endl;
		  ndbout << "3 \t=> Perform updates in short table" << endl;
		  ndbout << "4 \t=> Perform updates in long table" << endl;
		  ndbout << "5 \t=> Perform 50% lookups/50% updates in short table" << endl;
		  ndbout << "6 \t=> Perform 50% lookups/50% updates in long table" << endl;
		  ndbout << "7 \t=> Perform 80% lookups/20% updates in short table" << endl;
		  ndbout << "8 \t=> Perform 80% lookups/20% updates in long table" << endl;
		  ndbout << "9 \t=> Perform 25% lookups short/25% lookups long/25% updates short/25% updates long" << endl;
		  ndbout << "10\t=> Test bug with replicated interpreted updates, short table" << endl;
		  ndbout << "11\t=> Test interpreter functions, short table" << endl;
		  ndbout << "12\t=> Test bug with replicated interpreted updates, long table" << endl;
		  ndbout << "13\t=> Test interpreter functions, long table" << endl;
		  ndbout << "14\t=> Perform lookups in short table, no guess of TC" << endl;
		  ndbout << "15\t=> Perform lookups in long table, no guess of TC" << endl;
		  ndbout << "16\t=> Perform updates in short table, no guess of TC" << endl;
		  ndbout << "17\t=> Perform updates in long table, no guess of TC" << endl;
		  ndbout << "18\t=> Multi record updates of transactions" << endl;
		  ndbout << "All other responses will exit" << endl;
		  ndbout << "_____________________________" << endl << endl ;
		  
		  int inp_i = 0;
		  do {
		    inp[inp_i] = (char) fgetc(stdin);		    
		    if (inp[inp_i] == '\n' || inp[inp_i] == EOF) {
		      inp[inp_i] ='\0';
		      break;
		    }		
		    inp_i++;

		  } while (inp[inp_i - 1] != '\n' && inp[inp_i - 1] != EOF);
		  
		  tmp = atoi(inp);
		  
		  if ((tmp > 18) || (tmp <= 0)) break;
		  
		  ndbout << "Starting test " << tmp << "..." << endl;

		  for (i = 0; i < tNoOfThreads ; i++){ ThreadStart[i] = tmp; }
		  
		  cont = 1;
		  while (cont) {
			  NdbSleep_MilliSleep(10);
			  cont = 0;
			  for (i = 0; i < tNoOfThreads ; i++){
				  if (!ThreadReady[i]) cont = 1;
			  }
		  }//while
	  }//for(;;)

  }else{

	  if(19 == nTest){
		  ndbout << "Executing all 18 available tests..." << endl << endl;
		  for (int count = 1; count < nTest; count++){
			  ndbout << "Test " << count << endl ;
			  ndbout << "------" << endl << endl ;
			  for (i = 0; i < tNoOfThreads ; i++) { ThreadStart[i] = count ; }
			  cont = 1;
			  while (cont) {
				  NdbSleep_MilliSleep(10);
				  cont = 0;
				  for (i = 0; i < tNoOfThreads ; i++){
					  if (!ThreadReady[i]) cont = 1;
				  }
			  }
		  }//for
	  }else{
		  ndbout << endl << "Executing test " << nTest << endl << endl;
		  for (i = 0; i < tNoOfThreads ; i++) { ThreadStart[i] = nTest ; }
		  cont = 1;
		  while (cont) {
			  NdbSleep_MilliSleep(10);
			  cont = 0;
			  for (i = 0; i < tNoOfThreads ; i++){
				  if (!ThreadReady[i]) cont = 1;
			  }
		  }
	  }//if(18 == nTest)
  } //if(!nTest)

  ndbout << "--------------------------------------------------" << endl;

  for (i = 0; i < tNoOfThreads ; i++) ThreadReady[i] = 0;
  // Signaling threads to stop 
  for (i = 0; i < tNoOfThreads ; i++) ThreadStart[i] = 999;

    // Wait for threads to stop
  cont = 1;
  do {
     NdbSleep_MilliSleep(1);
	 cont = 0;
	 for (i = 0; i < tNoOfThreads ; i++){
      if (ThreadReady[i] == 0) cont = 1;
	 }
  } while (cont == 1);

  delete pMyNdb ;
  ndbout << endl << "Ronja Benchmark completed" << endl;
  return NDBT_ProgramExit(NDBT_OK) ;

error_input:
  ndbout << endl << "  Ivalid parameter(s)" << endl;
  ndbout <<         "  Usage: benchronja [-t threads][-r rec] [-o ops] [-c ops_per_exec] [-p test], where:" << endl;
  ndbout <<			"  threads - the number of threads to start; default: 1" << endl;
  ndbout <<			"  rec - the number of records in the tables; default: 500" << endl;
  ndbout <<			"  ops - the number of operations per transaction; default: 100000" << endl;
  ndbout <<			"  ops_per_exec - the number of operations per execution; default: 1" << endl ;
  ndbout <<			"  test - the number of test to execute; 19 executes all available tests; default: 0"<< endl ;
  ndbout <<			"  which enters a loop expecting manual input of test number to execute." << endl << endl ;
  delete pMyNdb ;
  return NDBT_ProgramExit(NDBT_WRONGARGS) ;

  }
////////////////////////////////////////

void commitTrans(Ndb* aNdb, NdbConnection* aCon)
{
  int ret = aCon->execute(Commit);
  assert (ret != -1);
  aNdb->closeTransaction(aCon);
}

void rollbackTrans(Ndb* aNdb, NdbConnection* aCon)
{
  int ret = aCon->execute(Rollback);
  assert (ret != -1);
  aNdb->closeTransaction(aCon);
}

void updateNoCommit(NdbConnection* aCon, Uint32* flip, unsigned int key)
{
  NdbOperation* theOperation;

  *flip = *flip + 1;
  theOperation = aCon->getNdbOperation("SHORT_REC");
  theOperation->updateTuple();
  theOperation->equal((Uint32)0, key);
  theOperation->setValue((Uint32)1, (char*)flip);
  int ret = aCon->execute(NoCommit);
  assert (ret != -1);
}

void updateNoCommitFail(NdbConnection* aCon, unsigned int key)
{
  NdbOperation* theOperation;

  Uint32 flip = 0;
  theOperation = aCon->getNdbOperation("SHORT_REC");
  theOperation->updateTuple();
  theOperation->equal((Uint32)0, key);
  theOperation->setValue((Uint32)1, (char*)flip);
  int ret = aCon->execute(NoCommit);
  assert (ret == -1);
}

void deleteNoCommit(NdbConnection* aCon, Uint32* flip, unsigned int key)
{
  NdbOperation* theOperation;

  *flip = 0;
  theOperation = aCon->getNdbOperation("SHORT_REC");
  theOperation->deleteTuple();
  theOperation->equal((Uint32)0, key);
  int ret = aCon->execute(NoCommit);
  assert (ret != -1);
}

void insertNoCommit(NdbConnection* aCon, Uint32* flip, unsigned int key)
{
  NdbOperation* theOperation;
  Uint32 placeholder[100];

  *flip = *flip + 1;
  theOperation = aCon->getNdbOperation("SHORT_REC");
  theOperation->insertTuple();
  theOperation->equal((Uint32)0, key);
  theOperation->setValue((Uint32)1, (char*)flip);
  theOperation->setValue((Uint32)2, (char*)&placeholder[0]);
  theOperation->setValue((Uint32)3, (char*)&placeholder[0]);
  int ret = aCon->execute(NoCommit);
  assert (ret != -1);
}

void writeNoCommit(NdbConnection* aCon, Uint32* flip, unsigned int key)
{
  NdbOperation* theOperation;
  Uint32 placeholder[100];

  *flip = *flip + 1;
  theOperation = aCon->getNdbOperation("SHORT_REC");
  theOperation->writeTuple();
  theOperation->equal((Uint32)0, key);
  theOperation->setValue((Uint32)1, (char*)flip);
  theOperation->setValue((Uint32)2, (char*)&placeholder[0]);
  theOperation->setValue((Uint32)3, (char*)&placeholder[0]);
  int ret = aCon->execute(NoCommit);
  assert (ret != -1);
}

void readNoCommit(NdbConnection* aCon, Uint32* flip, Uint32 key, int expected_ret)
{
  NdbOperation* theOperation;
  Uint32 readFlip;

  theOperation = aCon->getNdbOperation("SHORT_REC");
  theOperation->readTuple();
  theOperation->equal((Uint32)0, key);
  theOperation->getValue((Uint32)1, (char*)&readFlip);
  int ret = aCon->execute(NoCommit);
  assert (ret == expected_ret);
  if (ret == 0) 
    assert (*flip == readFlip);
}

void readDirtyNoCommit(NdbConnection* aCon, Uint32* flip, Uint32 key, int expected_ret)
{
  NdbOperation* theOperation;
  Uint32 readFlip;

  theOperation = aCon->getNdbOperation("SHORT_REC");
  theOperation->committedRead();
  theOperation->equal((Uint32)0, key);
  theOperation->getValue((Uint32)1, (char*)&readFlip);
  int ret = aCon->execute(NoCommit);
  assert (ret == expected_ret);
  if (ret == 0) 
    assert (*flip == readFlip);
}

void readVerify(Ndb* aNdb, Uint32* flip, Uint32 key, int expected_ret)
{
  NdbConnection* theTransaction;
  theTransaction = aNdb->startTransaction();
  readNoCommit(theTransaction, flip, key, expected_ret);
  commitTrans(aNdb, theTransaction);
}

void readDirty(Ndb* aNdb, Uint32* flip, Uint32 key, int expected_ret)
{
  NdbOperation* theOperation;
  NdbConnection* theTransaction;
  Uint32 readFlip;