/* 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 *//* ***************************************************       FLEXHAMMER       Hammer ndb with read, insert, update and delete transactions.        Arguments:        -t Number of threads to start, default 1        -o Number of operations per hammering-round, default 500	-l Number of loops to run, default 1, 0=infinite        -a Number of attributes, default 25        -c Number of tables, default 1	-s Size of each attribute, default 1	-simple Use simple read to read from database        -dirty Use dirty read to read from database	-write Use writeTuple to write to db	-r Number of records to Hammer        -no_table_create Don't create tables in db        -regulate To be able to regulate the load flexHammer produces.        -stdtables Use standard table names	-sleep Sleep a number of seconds before running the test, this 	       can be used so that another flexProgram have tome to create tables       Returns:        0 - Test passed       -1 - Test failed        1 - Invalid argumentsRevision history:  1.7  020208 epesson: Adapted to use NDBT  1.10 020222 epesson: Finalised handling of thread results  1.11 020222 epesson: Bug in checking results during delete fixed * *************************************************** */#include <ndb_global.h>#include <NdbApi.hpp>#include <NdbMain.h>#include <NdbThread.h>#include <NdbSleep.h>#include <NdbTick.h>#include <NdbOut.hpp>#include <NdbTimer.hpp>#include <NdbTick.h>#include <NdbTest.hpp>#include <NDBT_Error.hpp>#include <NdbSchemaCon.hpp>ErrorData * flexHammerErrorData;#if defined NDB_OSE || defined NDB_SOFTOSE#include <outfmt.h> #endif#define MAXSTRLEN 16 #define MAXATTR 64#define MAXTABLES 64#define MAXTHREADS 256#define MAXATTRSIZE 100// Max number of retries if something fails#define MaxNoOfAttemptsC 10 enum StartType {  stIdle,  stHammer,  stStop,  stLast};enum MyOpType {  otInsert,  otRead,  otDelete,  otUpdate,  otLast};struct ThreadNdb {  int threadNo;  NdbThread* threadLife;  int threadReady;  StartType threadStart;  int threadResult;};extern "C" void* flexHammerThread(void*);static int setAttrNames(void);static int setTableNames(void);static int readArguments(int, const char**);static int createTables(Ndb*);static void sleepBeforeStartingTest(int seconds);static int checkThreadResults(ThreadNdb *threadArrayP, char* phase);//enum OperationType {//  otInsert,//  otRead,//  otUpdate,//  otDelete,//  otVerifyDelete,//  otLast };enum ReadyType {	stReady, 	stRunning} ;static int			tNoOfThreads;static int			tNoOfAttributes;static int			tNoOfTables;static int			tNoOfBackups;static int			tAttributeSize;static int			tNoOfOperations;static int			tNoOfRecords;static int			tNoOfLoops;static				ReadyType ThreadReady[MAXTHREADS];static				StartType ThreadStart[MAXTHREADS];static char			tableName[MAXTABLES][MAXSTRLEN];static char			attrName[MAXATTR][MAXSTRLEN];static int			theSimpleFlag = 0;static int			theWriteFlag = 0;static int			theDirtyFlag = 0;static int			theTableCreateFlag = 0;static int			theStandardTableNameFlag = 0;static unsigned int tSleepTime = 0;#define START_TIMER { NdbTimer timer; timer.doStart();#define STOP_TIMER timer.doStop();#define PRINT_TIMER(text, trans, opertrans) timer.printTransactionStatistics(text, trans, opertrans); }; // Initialise thread datavoid resetThreads(ThreadNdb *threadArrayP) {  for (int i = 0; i < tNoOfThreads ; i++)    {      threadArrayP[i].threadReady = 0;      threadArrayP[i].threadResult = 0;      threadArrayP[i].threadStart = stIdle;    }} // resetThreadsvoid waitForThreads(ThreadNdb *threadArrayP){  int cont = 1;  while (cont) {    NdbSleep_MilliSleep(100);    cont = 0;    for (int i = 0; i < tNoOfThreads ; i++) {      if (threadArrayP[i].threadReady == 0) {	cont = 1;      } // if    } // for  } // while} // waitForThreadsvoid tellThreads(ThreadNdb* threadArrayP, const StartType what){  for (int i = 0; i < tNoOfThreads ; i++)     {    threadArrayP[i].threadStart = what;    } // for} // tellThreadsstatic Ndb_cluster_connection *g_cluster_connection= 0; NDB_COMMAND(flexHammer, "flexHammer", "flexHammer", "flexHammer", 65535)//main(int argc, const char** argv){  ndb_init();  ThreadNdb* pThreads = NULL; // Pointer to thread data array  Ndb* pMyNdb = NULL;	      // Pointer to Ndb object  int tLoops = 0;  int returnValue = 0;  int check = 0;    flexHammerErrorData = new ErrorData;  flexHammerErrorData->resetErrorCounters();  if (readArguments(argc, argv) != 0) {    ndbout << "Wrong arguments to flexHammer" << endl;    return NDBT_ProgramExit(NDBT_WRONGARGS);  } // if  /* print Setting */  flexHammerErrorData->printSettings(ndbout);  check = setAttrNames();  if (check == -1) {    ndbout << "Couldn't set attribute names" << endl;    return NDBT_ProgramExit(NDBT_FAILED);  } // if  check = setTableNames();  if (check == -1) {    ndbout << "Couldn't set table names" << endl;    return NDBT_ProgramExit(NDBT_FAILED);  } // if  // Create thread data array  pThreads = new ThreadNdb[tNoOfThreads];  // NdbThread_SetConcurrencyLevel(tNoOfThreads + 2);  // Create and init Ndb object  Ndb_cluster_connection con;  if(con.connect(12, 5, 1) != 0)  {    return NDBT_ProgramExit(NDBT_FAILED);  }  g_cluster_connection= &con;  pMyNdb = new Ndb(g_cluster_connection, "TEST_DB");  pMyNdb->init();  // Wait for Ndb to become ready  if (pMyNdb->waitUntilReady(10000) != 0) {    ndbout << "NDB is not ready" << endl << "Benchmark failed" << endl;    returnValue = NDBT_FAILED;  }  else {    check = createTables(pMyNdb);    if (check != 0) {      returnValue = NDBT_FAILED;    } // if    else {      sleepBeforeStartingTest(tSleepTime);            // Create threads.                                           *      resetThreads(pThreads);      for (int i = 0; i < tNoOfThreads ; i++) {  	pThreads[i].threadNo = i;	pThreads[i].threadLife = NdbThread_Create(flexHammerThread,						  (void**)&pThreads[i],						  65535,						  "flexHammerThread",                                                  NDB_THREAD_PRIO_LOW);      } // for            // And wait until they are ready      waitForThreads(pThreads);      if (checkThreadResults(pThreads, "init") != 0) {        returnValue = NDBT_FAILED;      } // if                  if (returnValue == NDBT_OK) {	ndbout << endl <<  "All threads started" << endl << endl;		for(;;) {	  	  // Check if it's time to exit program	  if((tNoOfLoops != 0) && (tNoOfLoops <= tLoops))	    break;	  	  // Tell all threads to start hammer	  ndbout << "Hammering..." << endl;	  	  resetThreads(pThreads);	  	  START_TIMER;	  tellThreads(pThreads, stHammer);	  	  waitForThreads(pThreads);	  ndbout << "Threads ready to continue..." << endl;	  STOP_TIMER;	  	  // Check here if anything went wrong	  if (checkThreadResults(pThreads, "hammer") != 0) {	    ndbout << "Thread(s) failed." << endl;	    returnValue = NDBT_FAILED;	  } // if	  	  PRINT_TIMER("hammer", tNoOfOperations*tNoOfThreads, tNoOfTables*6);	  	  ndbout << endl;	  	  tLoops++;	  	} // for      } // if      // Signaling threads to stop       resetThreads(pThreads);      tellThreads(pThreads, stStop);            // Wait for threads to stop      waitForThreads(pThreads);            ndbout << "----------------------------------------------" << endl << endl;      ndbout << "Benchmark completed" << endl;    } // else  } // else  // Clean up   flexHammerErrorData->printErrorCounters(ndbout);  // Kill them all!   void* tmp;  for(int i = 0; i < tNoOfThreads; i++){    NdbThread_WaitFor(pThreads[i].threadLife, &tmp);    NdbThread_Destroy(&pThreads[i].threadLife);  }  delete flexHammerErrorData;  delete [] pThreads;  delete pMyNdb;  // Exit via NDBT  return NDBT_ProgramExit(returnValue);  } //mainextern "C"void*flexHammerThread(void* pArg){  ThreadNdb* pThreadData = (ThreadNdb*)pArg;  unsigned int threadNo = pThreadData->threadNo;  Ndb* pMyNdb = NULL ;  NdbConnection *pMyTransaction = NULL ;  //  NdbOperation*	pMyOperation[MAXTABLES] = {NULL};  NdbOperation*	pMyOperation[MAXTABLES];  int check = 0;  int loop_count_ops = 0;  int loop_count_tables = 0;  int loop_count_attributes = 0;  int count_round = 0;  int count = 0;  int count_tables = 0;  int count_attributes = 0;  int i = 0;  int j = 0;  int tThreadResult = 0;  MyOpType tMyOpType = otLast;  int pkValue = 0;  int readValue[MAXATTR][MAXATTRSIZE] = {0};  int attrValue[MAXATTRSIZE];  NdbRecAttr* tTmp = NULL;  int tNoOfAttempts = 0;   for (i = 0; i < MAXATTRSIZE; i++)    attrValue[i] = 0;   // Ndb object for each thread  pMyNdb = new Ndb(g_cluster_connection, "TEST_DB" );  pMyNdb->init();  if (pMyNdb->waitUntilReady(10000) != 0) {    // Error, NDB is not ready    tThreadResult = 99;    // Go to idle directly    pThreadData->threadStart = stIdle;  } // if    for(;;) {    pThreadData->threadResult = tThreadResult;    pThreadData->threadReady = 1; // Signalling ready to main        // If Idle just wait to be stopped from main    while (pThreadData->threadStart == stIdle) {      NdbSleep_MilliSleep(100);       } // while        // Check if signal to exit is received    if (pThreadData->threadStart == stStop) {      pThreadData->threadReady = 1;      // break out of eternal loop      break;    } // if        // Set to Idle to prepare for possible error break    pThreadData->threadStart = stIdle;        // Prepare transaction    loop_count_ops = tNoOfOperations;    loop_count_tables = tNoOfTables;    loop_count_attributes = tNoOfAttributes;        for (count=0 ; count < loop_count_ops ; count++) {            //pkValue = (int)(count + thread_base);      // This limits the number of records used in this test      pkValue = count % tNoOfRecords;             for (count_round = 0; count_round < 5; ) {	switch (count_round) {	case 0:       // Insert	  tMyOpType = otInsert;	  // Increase attrValues	  for (i=0; i < MAXATTRSIZE; i ++) {	    attrValue[i]++;	  }	  break;	case 1: 	case 3:       // Read and verify	  tMyOpType = otRead;	  break;	case 2:       // Update	  // Increase attrValues	  for(i=0; i < MAXATTRSIZE; i ++) {	    attrValue[i]++;	  }	  tMyOpType = otUpdate;	  break;	case 4:       // Delete	  tMyOpType = otDelete;	  break;	default:	  assert(false);	  break;	} // switch	    	// Get transaction object	pMyTransaction = pMyNdb->startTransaction();	if (pMyTransaction == NULL) {	  // Fatal error	  tThreadResult = 1;	  // break out of for count_round loop waiting to be stopped by main	  break;	} // if	for (count_tables = 0; count_tables < loop_count_tables; 	     count_tables++) {	  pMyOperation[count_tables] = 	    pMyTransaction->getNdbOperation(tableName[count_tables]);		  if (pMyOperation[count_tables] == NULL) {	    //Fatal error	    tThreadResult = 2;	    // break out of inner for count_tables loop	    break;	  } // if			  switch (tMyOpType) {	  case otInsert:			// Insert case	    if (theWriteFlag == 1 && theDirtyFlag == 1) {	      check = pMyOperation[count_tables]->dirtyWrite();	    } else if (theWriteFlag == 1) {	      check = pMyOperation[count_tables]->writeTuple();