flex_bench_mysql.cpp

MySQL数据库开发源码 值得一看哦

/* 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
    -use_ndb Use NDB API, otherwise use mysql client
#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

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

#define USE_MYSQL
#ifdef USE_MYSQL
#include <mysql.h>
#endif

#include "NdbApi.hpp"

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

#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);
#ifdef USE_MYSQL
static int createTables(MYSQL*);
static int dropTables(MYSQL*);
#endif
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;
static unsigned int         t_instances = 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 bool                 verbose = false;
#ifdef USE_MYSQL
static bool                 use_ndb = false;
static int                  engine_id = 0;
static int                  sockets[16];
static int                  n_sockets = 0;
static char*                engine[] =
  { 
    " ENGINE = NDBCLUSTER ", // use default engine
    " ENGINE = MEMORY ",
    " ENGINE = MYISAM ",
    " ENGINE = INNODB "
  };
#else
static bool                 use_ndb = true;
#endif

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;
}

NDB_COMMAND(flexBench, "flexBench", "flexBench", "flexbench", 65535)
{
  ndb_init();
  ThreadData*           pThreadsData;
  int                   tLoops = 0;
  int                   returnValue = NDBT_OK;
  if (readArguments(argc, argv) != 0){
    input_error();
    return NDBT_ProgramExit(NDBT_WRONGARGS);
  }
  NdbAutoPtr<char> p10;
  if(useLongKeys){
    int e1 = sizeof(char*) * tNoOfLongPK;
    int e2_1 = strlen("KEYATTR  ") + 1;
    int e2 = e2_1 * tNoOfLongPK;
    char *tmp = (char *) malloc(e1 + e2);
    p10.reset(tmp);
    longKeyAttrName = (char **) tmp;
    tmp += e1;
    for (Uint32 i = 0; i < tNoOfLongPK; i++) {
      //      longKeyAttrName[i] = (char *) malloc(strlen("KEYATTR  ") + 1);
      longKeyAttrName[i] = tmp;
      tmp += e2_1;
      memset(longKeyAttrName[i], 0, e2_1);
      sprintf(longKeyAttrName[i], "KEYATTR%i", i);
    }
  }

  NdbAutoObjArrayPtr<ThreadData>
    p12( 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 ;
  }
  if (use_ndb) {
    ndbout << "Use NDB API with NdbPool in this test case" << endl;
    ndbout << "Pool size = " << t_instances << endl;
  } else {
    ndbout << "Use mysql client with " << engine[engine_id];
    ndbout << " as engine" << endl;
  }
  theErrorData.printSettings(ndbout);
  
  NdbThread_SetConcurrencyLevel(tNoOfThreads + 2);

#ifdef USE_MYSQL
  MYSQL mysql;
  if (!use_ndb) {
    if ( mysql_thread_safe() == 0 ) {
      ndbout << "Not thread safe mysql library..." << endl;
      return NDBT_ProgramExit(NDBT_FAILED);
    }

    ndbout << "Connecting to MySQL..." <<endl;

    mysql_init(&mysql);
    {
      int the_socket = sockets[0];
      char the_socket_name[1024];
      sprintf(the_socket_name, "%s%u%s", "/tmp/mysql.",the_socket,".sock");
      //    sprintf(the_socket_name, "%s", "/tmp/mysql.sock");
      ndbout << the_socket_name << endl;
      if ( mysql_real_connect(&mysql,
			    "localhost",
			    "root",
			    "",
			    "test",
			    the_socket,
			    the_socket_name,
			    0) == NULL ) {
        ndbout << "Connect failed" <<endl;
        returnValue = NDBT_FAILED;
      }
      mysql.reconnect= 1;
    }
    if(returnValue == NDBT_OK){
      mysql_set_server_option(&mysql, MYSQL_OPTION_MULTI_STATEMENTS_ON);
      if (createTables(&mysql) != 0){
        returnValue = NDBT_FAILED;
      }
    }
  }
#endif
  if (use_ndb) {
    Uint32 ndb_id = 0;
    if (!create_instance(t_instances, 1, t_instances)) {
      ndbout << "Creation of the NdbPool failed" << endl;
      returnValue = NDBT_FAILED;
    } else {
      Ndb* pNdb = get_ndb_object(ndb_id, "test", "def");
      if (pNdb == NULL) {
        ndbout << "Failed to get a NDB object" << endl;
        returnValue = NDBT_FAILED;
      } else {
        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;
          }
        }
        return_ndb_object(pNdb, ndb_id);
      }
    }