flexhammer.cpp

mysql-5.0.22.tar.gz源码包

/* 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 arguments

Revision 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 data
void 
resetThreads(ThreadNdb *threadArrayP) {

  for (int i = 0; i < tNoOfThreads ; i++)
    {
      threadArrayP[i].threadReady = 0;
      threadArrayP[i].threadResult = 0;
      threadArrayP[i].threadStart = stIdle;
    }
} // resetThreads

void 
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
} // waitForThreads

void 
tellThreads(ThreadNdb* threadArrayP, const StartType what)
{
  for (int i = 0; i < tNoOfThreads ; i++) 
    {
    threadArrayP[i].threadStart = what;
    } // for
} // tellThreads

static 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);
  
} //main

extern "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();