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


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

Arguments:
    -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 (Primary Key is always of size 1,
    independent of this value)
    -lkn Number of long primary keys, default 1
    -lks Size of each long primary key, default 1
    -simple Use simple read to read from database
    -dirty Use dirty read to read from database
    -write Use writeTuple in insert and update
    -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 have time to create tables
    -temp Use tables without logging
    -verify Verify inserts, updates and deletes
#ifdef CEBIT_STAT
    -statserv host:port  statistics server to report to
    -statfreq ops        report every ops operations (default 100)
#endif
    Returns:
    0 - Test passed
    1 - Test failed
    2 - Invalid arguments

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

#include <ndb_global.h>
#include "NdbApi.hpp"

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

#include <NdbTest.hpp>

#define MAXSTRLEN 16 
#define MAXATTR 64
#define MAXTABLES 128
#define MAXATTRSIZE 1000
#define MAXNOLONGKEY 16 // Max number of long keys.
#define MAXLONGKEYTOTALSIZE 1023 // words = 4092 bytes

extern "C" { static void* flexBenchThread(void*); }
static int readArguments(int argc, const char** argv);
static int createTables(Ndb*);
static void sleepBeforeStartingTest(int seconds);
static void input_error();

enum StartType { 
  stIdle,
  stInsert,
  stVerify,
  stRead,
  stUpdate,
  stDelete,
  stTryDelete,
  stVerifyDelete,
  stStop 
};

struct ThreadData
{
  int threadNo;
  NdbThread* threadLife;
  int threadReady;  
  StartType threadStart;
  int threadResult;
};

static int                  tNodeId = 0 ;
static char                 tableName[MAXTABLES][MAXSTRLEN+1];
static char                 attrName[MAXATTR][MAXSTRLEN+1];
static char**               longKeyAttrName;

// Program Parameters
static int                  tNoOfLoops = 1;
static int                  tAttributeSize = 1;
static unsigned int         tNoOfThreads = 1;
static unsigned int         tNoOfTables = 1;
static unsigned int         tNoOfAttributes = 25;
static unsigned int         tNoOfOperations = 500;
static unsigned int         tSleepTime = 0;
static unsigned int         tNoOfLongPK = 1;
static unsigned int         tSizeOfLongPK = 1;

//Program Flags
static int                  theSimpleFlag = 0;
static int                  theDirtyFlag = 0;
static int                  theWriteFlag = 0;
static int                  theStdTableNameFlag = 0;
static int                  theTableCreateFlag = 0;
static bool                 theTempTable = false;
static bool                 VerifyFlag = true;
static bool                 useLongKeys = false;


static ErrorData theErrorData; // Part of flexBench-program

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

#include <NdbTCP.h>

#ifdef CEBIT_STAT
#include <NdbMutex.h>
static bool statEnable = false;
static char statHost[100];
static int statFreq = 100;
static int statPort = 0;
static int statSock = -1;
static enum { statError = -1, statClosed, statOpen } statState;
static NdbMutex statMutex = NDB_MUTEX_INITIALIZER;
#endif

//-------------------------------------------------------------------
// Statistical Reporting routines
//-------------------------------------------------------------------
#ifdef CEBIT_STAT
// Experimental client-side statistic for CeBIT

static void
statReport(enum StartType st, int ops)
{
  if (!statEnable)
    return;
  if (NdbMutex_Lock(&statMutex) < 0) {
    if (statState != statError) {
      ndbout_c("stat: lock mutex failed: %s", strerror(errno));
      statState = statError;
    }
    return;
  }
  static int nodeid;
  // open connection
  if (statState != statOpen) {
    char *p = getenv("NDB_NODEID");		// ndbnet sets NDB_NODEID
    nodeid = p == 0 ? 0 : atoi(p);
    if ((statSock = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
      if (statState != statError) {
	ndbout_c("stat: create socket failed: %s", strerror(errno));
	statState = statError;
      }
      (void)NdbMutex_Unlock(&statMutex);
      return;
    }
    struct sockaddr_in saddr;
    memset(&saddr, 0, sizeof(saddr));
    saddr.sin_family = AF_INET;
    saddr.sin_port = htons(statPort);
    if (Ndb_getInAddr(&saddr.sin_addr, statHost) < 0) {
      if (statState != statError) {
	ndbout_c("stat: host %s not found", statHost);
	statState = statError;
      }
      (void)close(statSock);
      (void)NdbMutex_Unlock(&statMutex);
      return;
    }
    if (connect(statSock, (struct sockaddr *)&saddr, sizeof(saddr)) < 0) {
      if (statState != statError) {
	ndbout_c("stat: connect failed: %s", strerror(errno));
	statState = statError;
      }
      (void)close(statSock);
      (void)NdbMutex_Unlock(&statMutex);
      return;
    }
    statState = statOpen;
    ndbout_c("stat: connection to %s:%d opened", statHost, (int)statPort);
  }
  const char *text;
  switch (st) {
  case stInsert:
    text = "insert";
    break;
  case stVerify:
    text = "verify";
    break;
  case stRead:
    text = "read";
    break;
  case stUpdate:
    text = "update";
    break;
  case stDelete:
    text = "delete";
    break;
  case stVerifyDelete:
    text = "verifydelete";
    break;
  default:
    text = "unknown";
    break;
  }
  char buf[100];
  sprintf(buf, "%d %s %d\n", nodeid, text, ops);
  int len = strlen(buf);
  // assume SIGPIPE already ignored
  if (write(statSock, buf, len) != len) {
    if (statState != statError) {
      ndbout_c("stat: write failed: %s", strerror(errno));
      statState = statError;
    }
    (void)close(statSock);
    (void)NdbMutex_Unlock(&statMutex);
    return;
  }
  (void)NdbMutex_Unlock(&statMutex);
}
#endif	// CEBIT_STAT

static void 
resetThreads(ThreadData* pt){
  for (unsigned int i = 0; i < tNoOfThreads; i++){
    pt[i].threadReady = 0;
    pt[i].threadResult = 0;
    pt[i].threadStart = stIdle;
  }
}

static int 
checkThreadResults(ThreadData* pt){
  for (unsigned int i = 0; i < tNoOfThreads; i++){
    if(pt[i].threadResult != 0){
      ndbout_c("Thread%d reported fatal error %d", i, pt[i].threadResult);
      return -1;
    }
  }
  return 0;
}

static
void 
waitForThreads(ThreadData* pt)
{
  int cont = 1;
  while (cont){
    NdbSleep_MilliSleep(100);
    cont = 0;
    for (unsigned int i = 0; i < tNoOfThreads; i++){
      if (pt[i].threadReady == 0) 
    cont = 1;
    }
  }
}

static void 
tellThreads(ThreadData* pt, StartType what)
{
  for (unsigned int i = 0; i < tNoOfThreads; i++) 
    pt[i].threadStart = what;
}

static Ndb_cluster_connection *g_cluster_connection= 0;

NDB_COMMAND(flexBench, "flexBench", "flexBench", "flexbench", 65535)
{
  ndb_init();
  ThreadData*           pThreadsData;
  int                   tLoops = 0, i;
  int                   returnValue = NDBT_OK;
    
  if (readArguments(argc, argv) != 0){
    input_error();
    return NDBT_ProgramExit(NDBT_WRONGARGS);
  }

  if(useLongKeys){
    longKeyAttrName = (char **) malloc(sizeof(char*) * tNoOfLongPK);
    for (Uint32 i = 0; i < tNoOfLongPK; i++) {
      longKeyAttrName[i] = (char *) malloc(strlen("KEYATTR  ") + 1);
      memset(longKeyAttrName[i], 0, strlen("KEYATTR  ") + 1);
      sprintf(longKeyAttrName[i], "KEYATTR%i", i);
    }
  }
  
  pThreadsData = new ThreadData[tNoOfThreads];
 
  ndbout << endl << "FLEXBENCH - Starting normal mode" << endl;
  ndbout << "Perform benchmark of insert, update and delete transactions"<< 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 " << endl;
  ndbout << "  " << tNoOfOperations << " transaction(s) per thread and round " << endl;
  ndbout << "  " << tAttributeSize << " is the number of 32 bit words per attribute "<< endl;
  ndbout << "  " << "Table(s) without logging: " << (Uint32)theTempTable << endl;
  
  if(useLongKeys)
    ndbout << "  " << "Using long keys with " << tNoOfLongPK << " keys a' " << 
      tSizeOfLongPK * 4 << " bytes each." << endl;
  
  ndbout << "  " << "Verification is " ; 
  if(VerifyFlag) {
      ndbout << "enabled" << endl ;
  }else{
      ndbout << "disabled" << endl ;
  }
  theErrorData.printSettings(ndbout);
  
  NdbThread_SetConcurrencyLevel(tNoOfThreads + 2);

  Ndb_cluster_connection con;
  if(con.connect(12, 5, 1) != 0)
  {
    return NDBT_ProgramExit(NDBT_FAILED);
  }

  g_cluster_connection= &con;

  Ndb* pNdb;
  pNdb = new Ndb(&con, "TEST_DB" );  
  pNdb->init();

  tNodeId = pNdb->getNodeId();
  ndbout << "  NdbAPI node with id = " << tNodeId << endl;
  ndbout << endl;
  
  ndbout << "Waiting for ndb to become ready..." <<endl;
  if (pNdb->waitUntilReady(2000) != 0){
    ndbout << "NDB is not ready" << endl;
    ndbout << "Benchmark failed!" << endl;
    returnValue = NDBT_FAILED;
  }

  if(returnValue == NDBT_OK){
    if (createTables(pNdb) != 0){
      returnValue = NDBT_FAILED;
    }
  }

  if(returnValue == NDBT_OK){

    sleepBeforeStartingTest(tSleepTime);
    
    /****************************************************************
     *  Create threads.                                           *
     ****************************************************************/
    resetThreads(pThreadsData);
    
    for (i = 0; i < tNoOfThreads; i++){  
      pThreadsData[i].threadNo = i;
      pThreadsData[i].threadLife = NdbThread_Create(flexBenchThread,
                                                    (void**)&pThreadsData[i],
                                                    32768,
                                                    "flexBenchThread",
                                                    NDB_THREAD_PRIO_LOW);
    }
    
    waitForThreads(pThreadsData);
    
    ndbout << endl <<  "All threads started" << endl << endl;
    
    /****************************************************************
     * Execute program.                                             *
     ****************************************************************/
  
    for(;;){

      int loopCount = tLoops + 1;