flexscan.cpp

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

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

/* ***************************************************
       FLEXSCAN
       Perform benchmark of:
         insert
	 read
	 scan read
	 update
	 scan update
	 read
	 scan delete
	 verify delete

       Arguments:
        -f Location of Ndb.cfg file, default Ndb.cfg
        -t Number of threads to start, default 1
        -o Number of operations per loop, 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
	-stdtables Use standard table names
        -no_table_create Don't create tables in db
	-sleep Sleep a number of seconds before running the test, this 
	       can be used so that another flexBench hav etome to create tables
	-p Parallellism to use 1-32, default:1
	-abort <number> Test scan abort after a number of tuples
	-h Print help text
	-no_scan_update Don't do scan updates
	-no_scan_delete Don't do scan deletes
	
       Returns:
        NDBT_OK - Test passed
        NDBT_FAILED - Test failed

  Revision history:
    1.12 020222 epesson: Rewritten to use NDBT. Major bugs fixed

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

#include "NdbApi.hpp"

#include <NdbThread.h>
#include <NdbSleep.h>
#include <NdbTick.h>
#include <NdbOut.hpp>
#include <NdbTimer.hpp>
#include <NdbMain.h>
#include <NdbTest.hpp>
#include <NDBT_Error.hpp>
#include <NdbSchemaCon.hpp>

#define PKSIZE 1
#define FOREVER 1
#define MAXSTRLEN 16 
#define MAXATTR 64
#define MAXTABLES 64
#define MAXTHREADS 256
#define MAXATTRSIZE 64

enum StartType { 
  stIdle, 
  stInsert,
  stRead, 
  stScanRead,
  stUpdate,
  stScanUpdate,
  stDelete,
  stVerifyDelete, 
  stScanDelete,
  stStop,
  stLast} ;


struct ThreadNdb
{
  int ThreadNo;
  NdbThread* threadLife;
  StartType threadStart;
  int threadResult;
  int threadReady;
};

extern "C" void* flexScanThread(void*);
static int setAttrNames(void);
static int setTableNames(void);
static int createTables(Ndb* pMyNdb);
static void sleepBeforeStartingTest(int seconds);
static int readArguments(int argc, const char** argv);
static void setAttrValues(int* attrValue, 
			  int* readValue, 
			  int Offset);
static int insertRows(Ndb* pNdb, int* pkValue, int* attrValue, StartType tType);
static int readRows(Ndb* pNdb, int* pkValue, int* readValue);
static int deleteRows(Ndb* pNdb, int* pkValue);
static int scanReadRows(Ndb* pNdb, int* readValue);
static int scanUpdateRows(Ndb* pNdb, int* readValue, int* attrValue);
static int scanDeleteRows(Ndb* pNdb, int* readValue);
static int verifyDeleteRows(Ndb* pNdb, int* pkValue, int* readValue);
static void Compare(int* attrValue, int* readValue);
static void UpdateArray(int *attrValue);

static int tNoOfThreads = 1;
static int tNoOfAttributes = 25;
static int tNoOfTables = 1;
static int tAttributeSize = 1;
static int tNodeId = 0;
static int tNoOfOperations = 500;
static int tNoOfLoops = 1;
static int tAbortAfter = 0;
static int tParallellism = 1;

static char tableName[MAXTABLES][MAXSTRLEN];
static char attrName[MAXATTR][MAXSTRLEN];

static unsigned int tSleepTime = 0;

static int theStdTableNameFlag = 0;
static int theTableCreateFlag = 0;
static int theScanAbortTestFlag = 0;
static int theNoScanUpdateFlag = 0;
static int theNoScanDeleteFlag = 0;

//flexScanErrorData = new ErrorData;
ErrorData * flexScanErrorData;
NdbError  * anerror;

//static errorData            theErrorData; 
//static unsigned int         tErrorCounter[6000]; 

#define START_TIMER { NdbTimer timer; timer.doStart();
#define STOP_TIMER timer.doStop();
#define PRINT_TIMER(text, trans, opertrans) timer.printTransactionStatistics(text, trans, opertrans); }; 

static void UpdateArray(int *attrValue)
{
  int tableCount = 0;
  int attrCount = 0;
  int opCount = 0;
  int sizeCount = 0;
  int* pValue = attrValue;

  for (tableCount = 0; tableCount < tNoOfTables; tableCount++) {
    for (attrCount = 0; attrCount < tNoOfAttributes-1; attrCount++) {
      for (opCount = 0; opCount < tNoOfOperations; opCount++) {
	for (sizeCount = 0; sizeCount < tAttributeSize; sizeCount++) {
	  // Update value in array
	  (*pValue)++;
	  //ndbout << "attrValue[" << tableCount*tNoOfAttributes*tNoOfOperations*tAttributeSize +
	  //attrCount*tNoOfOperations*tAttributeSize + opCount*tAttributeSize + sizeCount <<
	  //"] = " << attrValue[tableCount*tNoOfAttributes*tNoOfOperations*tAttributeSize +
	  //attrCount*tNoOfOperations*tAttributeSize + opCount*tAttributeSize + sizeCount] << endl;
	  // Increment pointer
	  pValue++;
	} // sizeCount
      } // for opCount
    } // for attrCount
  } // for tableCount

} // Update

static void Compare(int* attrValue, int* readValue)
{
  int tableCount = 0;
  int attrCount = 0;
  int OpCount = 0;
  int first = 0;

  for (tableCount = 0; tableCount < tNoOfTables; tableCount++) {
    for (attrCount = 0; attrCount < tNoOfAttributes-1; attrCount++) {
      for (OpCount = 0; OpCount < tNoOfOperations; OpCount++) {
	if (memcmp(&(attrValue[tableCount*(tNoOfAttributes-1)*tNoOfOperations + 
			      attrCount*tNoOfOperations + OpCount]),
		   &(readValue[tableCount*(tNoOfAttributes-1)*tNoOfOperations + 
			      attrCount*tNoOfOperations + OpCount]), 
		   tAttributeSize) != 0) {
	  // Values mismatch
	  if (first == 0) {
	    //ndbout << "Read and set values differ for:" << endl;
	    first = 1;
	    ndbout << "Mismatch found.";
	  } // if
	  // Comparision of values after scan update is meaningless right now
	  //ndbout << "  table " << tableName[tableCount] << 
	  //" - attr " << attrName[attrCount+1];
	  //for (sizeCount = 0; sizeCount < tAttributeSize; sizeCount++) {
	  //ndbout << ": set " << 
	  //attrValue[tableCount*(tNoOfAttributes-1)*tNoOfOperations*tAttributeSize + 
	  //attrCount*tNoOfOperations*tAttributeSize + 
	  //tNoOfOperations*tAttributeSize + sizeCount] << " read " << 
	  //readValue[tableCount*(tNoOfAttributes-1)*tNoOfOperations*tAttributeSize + 
	  //attrCount*tNoOfOperations*tAttributeSize + 
	  //tNoOfOperations*tAttributeSize + sizeCount] << endl;
	  //} // for
	} // if
      } // for OpCount
    } // for attrCount
  } // for tableCount

  // A final pretty-print
  if (first == 1) {
    ndbout << endl;
  } // if
} // Compare

static void printInfo()
{
  ndbout << endl << "FLEXSCAN - Starting normal mode" << endl;
  ndbout << "Perform benchmark of insert, update and delete transactions"<< endl;
  ndbout << "  NdbAPI node with id = " << tNodeId << endl;
  ndbout << "  " << tNoOfThreads << " thread(s) " << endl;
  ndbout << "  " << tNoOfLoops << " iterations " << endl;
  ndbout << "  " << tNoOfTables << " table(s) and " << 1 << " operation(s) per transaction " 
	 << endl;
  ndbout << "  " << tNoOfAttributes << " attributes per table incl. pk" << endl;
  ndbout << "  " << tNoOfOperations << " transaction(s) per thread and round " << endl;
  if (theScanAbortTestFlag == 1) {
    ndbout << "  Scan abort test after " << tAbortAfter << " tuples" << endl;
  } // if
  ndbout << "  " << tParallellism << " parallellism in scans" << endl;
  ndbout << "  " << tAttributeSize << " is the number of 32 bit words per attribute " << 
    endl << endl;
  
} // printInfo

static void tellThreads(ThreadNdb *threadArrayP, StartType what)
{
  int i = 0;

  for (i = 0; i < tNoOfThreads ; i++)
    threadArrayP[i].threadStart = what;
} // tellThreads

static void waitForThreads(ThreadNdb *threadArrayP)
{
  int i = 0;
  int cont = 1;

  while (cont == 1){

    NdbSleep_MilliSleep(10);
    cont = 0;
    
    for (i = 0; i < tNoOfThreads ; i++) {
      if (threadArrayP[i].threadReady == 0) {
//	ndbout << "Main is reporting thread " << i << " not ready" << endl;
	cont = 1;
      } // if
    } // for
  } // while
} // waitForThreads


static void resetThreads(ThreadNdb *threadArrayP)
{
  int i = 0;

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

static int checkThreadResults(ThreadNdb *threadArrayP, char *action)
{
  int i = 0;
  int retValue = 0;

  for (i = 0; i < tNoOfThreads; i++) {
    if (threadArrayP[i].threadResult != 0) {
      ndbout << "Thread " << i << " reported fatal error " 
	     << threadArrayP[i].threadResult << " during " << action << endl;
      retValue = -1;
      break;
    } // if
  } // for

  return(retValue);
} // checkThreadResults

NDB_COMMAND(flexScan, "flexScan", "flexScan", "flexScan", 65535)
{
  ndb_init();
  ThreadNdb*		pThreads = NULL;
  Ndb*			pMyNdb = NULL;	
  int                   tLoops = 0;
  int                   check = 0;
  int                   returnValue = NDBT_OK;
  int                   every2ndScanDelete = 0; // Switch between scan delete and normal delete

  flexScanErrorData = new ErrorData;

  flexScanErrorData->resetErrorCounters();

  if (readArguments(argc, argv) != 0) {
    ndbout << "Wrong arguments to flexScan" << endl;
    return NDBT_ProgramExit(NDBT_WRONGARGS);
  } // if

  /* print Setting */
  flexScanErrorData->printSettings(ndbout);

  check = setAttrNames();
  if (check != 0) {
    ndbout << "Couldn't set attribute names" << endl;
    return NDBT_ProgramExit(NDBT_FAILED);
  } // if
  check = setTableNames();
  if (check != 0) {
    ndbout << "Couldn't set table names" << endl;
    return NDBT_ProgramExit(NDBT_FAILED);
  } // if

  pMyNdb = new Ndb ("TEST_DB");
  pMyNdb->init();
  tNodeId = pMyNdb->getNodeId();

  printInfo();
                                  
  NdbThread_SetConcurrencyLevel(tNoOfThreads + 2);
  //NdbThread_SetConcurrencyLevel(tNoOfThreads + 8);
  
  pThreads = new ThreadNdb[tNoOfThreads];

  if (pMyNdb->waitUntilReady(10000) != 0) {
    ndbout << "NDB is not ready" << endl << "Benchmark failed"	 << endl;
    returnValue = NDBT_FAILED;
  } // if
  
  else {
    
    if (createTables(pMyNdb) != 0) {
      ndbout << "Could not create tables" << endl;
      returnValue = NDBT_FAILED;
    } // if
    else {
      sleepBeforeStartingTest(tSleepTime);
      
      resetThreads(pThreads);
      // Create threads
      for (int i = 0; i < tNoOfThreads ; i++){  
	pThreads[i].ThreadNo = i;
	// Ignore the case that thread creation may fail
	pThreads[i].threadLife = NdbThread_Create(flexScanThread,
						  (void**)&pThreads[i],
						  327680,
						  "flexScanThread", NDB_THREAD_PRIO_LOW);
	if (pThreads[i].threadLife == NULL) {
	  ndbout << "Could not create thread " << i << endl;
	  returnValue = NDBT_FAILED;
	  // Use the number of threads that were actually created
	  tNoOfThreads = i;
	  break; // break for loop
	} // if
      } // for
      
      waitForThreads(pThreads);
      if (checkThreadResults(pThreads, "init") != 0) {
	returnValue = NDBT_FAILED;
      } // if
      
      if (returnValue == NDBT_OK) {
	ndbout << "All threads started" << endl;
	
	while (FOREVER) {
	  
	  resetThreads(pThreads);
	  
	  if ((tNoOfLoops != 0) && (tNoOfLoops <= tLoops)) {
	    break;
	  } // if
	  
	  // Insert
	  START_TIMER;
	  
	  tellThreads(pThreads, stInsert);
	  waitForThreads(pThreads);
	  
	  STOP_TIMER;
	  if (checkThreadResults(pThreads, "insert") != 0) {
	    returnValue = NDBT_FAILED;
	    break;
	  } // if
	  PRINT_TIMER("insert", tNoOfOperations*tNoOfThreads, tNoOfTables);
	  
	  resetThreads(pThreads);
	  
	  // Read
	  START_TIMER;
	  
	  tellThreads(pThreads, stRead);
	  waitForThreads(pThreads);
	  
	  STOP_TIMER;
	  if (checkThreadResults(pThreads, "read") != 0) {
	    returnValue = NDBT_FAILED;
	    break;
	  } // if
	  PRINT_TIMER("read", tNoOfOperations*tNoOfThreads, tNoOfTables);