flex_bench_mysql.cpp

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

	tSpecialTrans = 0;
	if (check == -1) {
	  if ((stVerifyDelete == tType) && 
	      (626 == pTrans->getNdbError().code)) {
	    // ----------------------------------------------
	    // It's good news - the deleted tuple is gone, 
	    // so reset "check" flag
	    // ----------------------------------------------
	    check = 0 ;
	  } else {
	    int retCode = 
	      theErrorData.handleErrorCommon(pTrans->getNdbError());
	    if (retCode == 1) {
	      ndbout_c("execute: %d, %d, %s", count, tType, 
		       pTrans->getNdbError().message );
	      ndbout_c("Error code = %d", pTrans->getNdbError().code );
	      tResult = 20;
	    } else if (retCode == 2) {
	      ndbout << "4115 should not happen in flexBench" << endl;
	      tResult = 20;
	    } else if (retCode == 3) {
// --------------------------------------------------------------------
// We are not certain if the transaction was successful or not.
// We must reexecute but might very well find that the transaction
// actually was updated. Updates and Reads are no problem here. Inserts
// will not cause a problem if error code 630 arrives. Deletes will
// not cause a problem if 626 arrives.
// --------------------------------------------------------------------
	      if ((tType == stInsert) || (tType == stDelete)) {
		tSpecialTrans = 1;
	      }//if
	    }//if
	  }//if
	}//if
	// Check if retries should be made
	if (check == -1 && tResult == 0) {
	  if (tAttemptNo < tRetryAttempts){
	    tAttemptNo++;
	  } else {
// --------------------------------------------------------------------
// Too many retries have been made, report error and break out of loop
// --------------------------------------------------------------------
	    ndbout << "Thread" << threadNo;
	    ndbout << ": too many errors reported" << endl;
	    tResult = 10;
	    break;
	  }//if            
	}//if
      }

      if (check == 0){
	// Go to the next record
	count++;
	tAttemptNo = 0;
#ifdef CEBIT_STAT
	// report successful ops
	if (statEnable) {
	  statOps += loopCountTables;
	  if (statOps >= statFreq) {
	    statReport(tType, statOps);
	    statOps = 0;
	  }//if
	}//if
#endif
      }//if

      if (stVerify == tType && 0 == check){
	int nTableOffset = 0 ;
	for (int a = 1 ; a < loopCountAttributes ; a++){
	  for (int tables = 0 ; tables < loopCountTables ; tables++){
	    nTableOffset = tables*loopCountAttributes*tAttributeSize;
	    int ov =*(int*)&attrValue[nTableOffset + tAttributeSize*a];
	    int nv =*(int*)&tmpAttrRefValue[tAttributeSize*a];
	    if (ov != nv){
	      ndbout << "Error in verify ";
	      ndbout << "pk = " << tmpAttrRefValue[0] << ":" << endl;
	      ndbout << "attrValue[" << nTableOffset + tAttributeSize*a << "] = " << ov << endl ;
	      ndbout << "attrRefValue[" << nRefLocalOpOffset + tAttributeSize*a << "]" << nv << endl ;
	      tResult = 11 ;
	      break ;
	    }//if
	  }//for
	}//for
      }// if(stVerify ... )
      if (use_ndb) {
	pNdb->closeTransaction(pTrans);
        return_ndb_object(pNdb, ndb_id);
        pNdb = NULL;
      }
    }// operations loop
#ifdef CEBIT_STAT
    // report remaining successful ops
    if (statEnable) {
      if (statOps > 0) {
	statReport(tType, statOps);
	statOps = 0;
      }//if
    }//if
#endif
    if (pNdb) {
      pNdb->closeTransaction(pTrans);
      return_ndb_object(pNdb, ndb_id);
      pNdb = NULL;
    }
  }

#ifdef USE_MYSQL
  if (!use_ndb) {
    mysql_close(&mysql);
    for (Uint32 i = 0; i < tNoOfTables; i++) {
      mysql_stmt_close(prep_insert[i]);
      mysql_stmt_close(prep_update[i]);
      mysql_stmt_close(prep_delete[i]);
      mysql_stmt_close(prep_read[i]);
    }
  }
#endif
  if (use_ndb && pNdb) {
    ndbout << "I got here " << endl;
    return_ndb_object(pNdb, ndb_id);
  }
  return NULL;
}


static int readArguments(int argc, const char** argv)
{

  int i = 1;
  while (argc > 1){
    if (strcmp(argv[i], "-t") == 0){
      tNoOfThreads = atoi(argv[i+1]);
      if ((tNoOfThreads < 1)) 
        return -1;
      argc -= 1;
      i++;
    }else if (strcmp(argv[i], "-o") == 0){
      tNoOfOperations = atoi(argv[i+1]);
      if (tNoOfOperations < 1) 
        return -1;;
      argc -= 1;
      i++;
    }else if (strcmp(argv[i], "-a") == 0){
      tNoOfAttributes = atoi(argv[i+1]);
      if ((tNoOfAttributes < 2) || (tNoOfAttributes > MAXATTR)) 
        return -1;
      argc -= 1;
      i++;
    }else if (strcmp(argv[i], "-c") == 0){
      tNoOfTables = atoi(argv[i+1]);
      if ((tNoOfTables < 1) || (tNoOfTables > MAXTABLES)) 
        return -1;
      argc -= 1;
      i++;
    }else if (strcmp(argv[i], "-stdtables") == 0){
      theStdTableNameFlag = 1;
    }else if (strcmp(argv[i], "-l") == 0){
      tNoOfLoops = atoi(argv[i+1]);
      if ((tNoOfLoops < 0) || (tNoOfLoops > 100000)) 
        return -1;
      argc -= 1;
      i++;
    }else if (strcmp(argv[i], "-pool_size") == 0){
      t_instances = atoi(argv[i+1]);
      if ((t_instances < 1) || (t_instances > 240)) 
        return -1;
      argc -= 1;
      i++;
#ifdef USE_MYSQL
    }else if (strcmp(argv[i], "-engine") == 0){
      engine_id = atoi(argv[i+1]);
      if ((engine_id < 0) || (engine_id > 3)) 
        return -1;
      argc -= 1;
      i++;
    }else if (strcmp(argv[i], "-socket") == 0){
      sockets[n_sockets] = atoi(argv[i+1]);
      if (sockets[n_sockets] <= 0)
        return -1;
      n_sockets++;
      argc -= 1;
      i++;
    }else if (strcmp(argv[i], "-use_ndb") == 0){
      use_ndb = true;
#endif
    }else if (strcmp(argv[i], "-s") == 0){
      tAttributeSize = atoi(argv[i+1]);
      if ((tAttributeSize < 1) || (tAttributeSize > MAXATTRSIZE)) 
        return -1;
      argc -= 1;
      i++;
    }else if (strcmp(argv[i], "-lkn") == 0){
     tNoOfLongPK = atoi(argv[i+1]);
     useLongKeys = true;
      if ((tNoOfLongPK < 1) || (tNoOfLongPK > MAXNOLONGKEY) || 
	  (tNoOfLongPK * tSizeOfLongPK) > MAXLONGKEYTOTALSIZE){
      	ndbout << "Argument -lkn is not in the proper range." << endl;  
	return -1;
      }
      argc -= 1;
      i++;
    }else if (strcmp(argv[i], "-lks") == 0){
      tSizeOfLongPK = atoi(argv[i+1]);
      useLongKeys = true;
      if ((tSizeOfLongPK < 1) || (tNoOfLongPK * tSizeOfLongPK) > MAXLONGKEYTOTALSIZE){
	ndbout << "Argument -lks is not in the proper range 1 to " << 
	  MAXLONGKEYTOTALSIZE << endl;
        return -1;
      }
      argc -= 1;
      i++;
    }else if (strcmp(argv[i], "-simple") == 0){
      theSimpleFlag = 1;
    }else if (strcmp(argv[i], "-write") == 0){
      theWriteFlag = 1;
    }else if (strcmp(argv[i], "-dirty") == 0){
      theDirtyFlag = 1;
    }else if (strcmp(argv[i], "-sleep") == 0){
      tSleepTime = atoi(argv[i+1]);
      if ((tSleepTime < 1) || (tSleepTime > 3600)) 
        return -1;
      argc -= 1;
      i++;
    }else if (strcmp(argv[i], "-no_table_create") == 0){
      theTableCreateFlag = 1;
    }else if (strcmp(argv[i], "-temp") == 0){
      theTempTable = true;
    }else if (strcmp(argv[i], "-noverify") == 0){
      VerifyFlag = false ;
    }else if (theErrorData.parseCmdLineArg(argv, i) == true){
      ; //empty, updated in errorArg(..)
    }else if (strcmp(argv[i], "-verify") == 0){
      VerifyFlag = true ;
#ifdef CEBIT_STAT
    }else if (strcmp(argv[i], "-statserv") == 0){
      if (! (argc > 2))
	return -1;
      const char *p = argv[i+1];
      const char *q = strrchr(p, ':');
      if (q == 0)
	return -1;
      BaseString::snprintf(statHost, sizeof(statHost), "%.*s", q-p, p);
      statPort = atoi(q+1);
      statEnable = true;
      argc -= 1;
      i++;
    }else if (strcmp(argv[i], "-statfreq") == 0){
      if (! (argc > 2))
	return -1;
      statFreq = atoi(argv[i+1]);
      if (statFreq < 1)
	return -1;
      argc -= 1;
      i++;
#endif
    }else{       
      return -1;
    }
    argc -= 1;
    i++;
  }
#ifdef USE_MYSQL
  if (n_sockets == 0) {
    n_sockets = 1;
    sockets[0] = 3306;
  }
#endif
  return 0;
}

static void sleepBeforeStartingTest(int seconds){
  if (seconds > 0){
      ndbout << "Sleeping(" <<seconds << ")...";
      NdbSleep_SecSleep(seconds);
      ndbout << " done!" << endl;
    }
}


#ifdef USE_MYSQL
static int
dropTables(MYSQL* mysqlp){
  char buf[2048];
  for(unsigned i = 0; i < tNoOfTables; i++){
    int pos = 0;
    ndbout << "Dropping " << tableName[i] << "... ";
    pos += sprintf(buf+pos, "%s", "DROP TABLE ");
    pos += sprintf(buf+pos, "%s%s", tableName[i], ";");
    if (verbose)
      ndbout << endl << buf << endl;
    if (mysql_query(mysqlp, buf) != 0){
      ndbout << "Failed!"<<endl
	     <<mysql_error(mysqlp)<<endl
	     <<buf<<endl;
    } else
      ndbout << "OK!" << endl;
  }
  
  return 0;
}
#endif

#ifdef USE_MYSQL
static int
createTables(MYSQL* mysqlp){

  for (Uint32 i = 0; i < tNoOfAttributes; i++){
    BaseString::snprintf(attrName[i], MAXSTRLEN, "COL%d", i);
  }

  // Note! Uses only uppercase letters in table name's
  // so that we can look at the tables with SQL
  for (Uint32 i = 0; i < tNoOfTables; i++){
    if (theStdTableNameFlag == 0){
      BaseString::snprintf(tableName[i], MAXSTRLEN, "TAB%d_%d", i, 
	       (int)(NdbTick_CurrentMillisecond() / 1000));
    } else {
      BaseString::snprintf(tableName[i], MAXSTRLEN, "TAB%d", i);
    }
  }
  
  char buf[2048];
  for(unsigned i = 0; i < tNoOfTables; i++){
    int pos = 0;
    ndbout << "Creating " << tableName[i] << "... ";
    
    pos += sprintf(buf+pos, "%s", "CREATE TABLE ");
    pos += sprintf(buf+pos, "%s%s", tableName[i], " ");
    if(useLongKeys){
      for(Uint32 i = 0; i < tNoOfLongPK; i++) {
      }
    } else {
      pos += sprintf(buf+pos, "%s%s%s",
		     "(", attrName[0], " int unsigned primary key");
    }
    for (unsigned j = 1; j < tNoOfAttributes; j++)
      pos += sprintf(buf+pos, "%s%s%s", ",", attrName[j], " int unsigned");
    pos += sprintf(buf+pos, "%s%s%s", ")", engine[engine_id], ";");
    if (verbose)
      ndbout << endl << buf << endl;
    if (mysql_query(mysqlp, buf) != 0)
      return -1;
    ndbout << "done" << endl;
  }
  return 0;
}
#endif

static int
createTables(Ndb* pMyNdb){

  for (Uint32 i = 0; i < tNoOfAttributes; i++){
    BaseString::snprintf(attrName[i], MAXSTRLEN, "COL%d", i);
  }

  // Note! Uses only uppercase letters in table name's
  // so that we can look at the tables with SQL
  for (Uint32 i = 0; i < tNoOfTables; i++){
    if (theStdTableNameFlag == 0){
      BaseString::snprintf(tableName[i], MAXSTRLEN, "TAB%d_%d", i, 
	       (int)(NdbTick_CurrentMillisecond() / 1000));
    } else {
      BaseString::snprintf(tableName[i], MAXSTRLEN, "TAB%d", i);
    }
  }
  
  for(unsigned i = 0; i < tNoOfTables; i++){
    ndbout << "Creating " << tableName[i] << "... ";
    
    NdbDictionary::Table tmpTable(tableName[i]);
    
    tmpTable.setStoredTable(!theTempTable);

    if(useLongKeys){
      for(Uint32 i = 0; i < tNoOfLongPK; i++) {
	NdbDictionary::Column col(longKeyAttrName[i]);
	col.setType(NdbDictionary::Column::Unsigned);
	col.setLength(tSizeOfLongPK);
	col.setPrimaryKey(true);
	tmpTable.addColumn(col);
      }
    } else {
      NdbDictionary::Column col(attrName[0]);
      col.setType(NdbDictionary::Column::Unsigned);
      col.setLength(1);
      col.setPrimaryKey(true);
      tmpTable.addColumn(col);
    }
    NdbDictionary::Column col;
    col.setType(NdbDictionary::Column::Unsigned);
    col.setLength(tAttributeSize);
    for (unsigned j = 1; j < tNoOfAttributes; j++){
      col.setName(attrName[j]);
      tmpTable.addColumn(col);
    }
    if(pMyNdb->getDictionary()->createTable(tmpTable) == -1){
      return -1;
    }
    ndbout << "done" << endl;
  }
  
  return 0;
}

      
static void input_error(){
  ndbout << endl << "Invalid argument!" << endl;
  ndbout << endl << "Arguments:" << endl;
  ndbout << "   -t Number of threads to start, default 1" << endl;
  ndbout << "   -o Number of operations per loop, default 500" << endl;
  ndbout << "   -l Number of loops to run, default 1, 0=infinite" << endl;
  ndbout << "   -a Number of attributes, default 25" << endl;
  ndbout << "   -c Number of tables, default 1" << endl;
  ndbout << "   -s Size of each attribute, default 1 (Primary Key is always of size 1," << endl;
  ndbout << "      independent of this value)" << endl;
  ndbout << "   -lkn Number of long primary keys, default 1" << endl;
  ndbout << "   -lks Size of each long primary key, default 1" << endl;

  ndbout << "   -simple Use simple read to read from database" << endl;
  ndbout << "   -dirty Use dirty read to read from database" << endl;
  ndbout << "   -write Use writeTuple in insert and update" << endl;
  ndbout << "   -stdtables Use standard table names" << endl;
  ndbout << "   -no_table_create Don't create tables in db" << endl;
  ndbout << "   -sleep Sleep a number of seconds before running the test, this" << endl;
  ndbout << "    can be used so that another flexBench have time to create tables" << endl;
  ndbout << "   -temp Use tables without logging" << endl;
  ndbout << "   -verify Verify inserts, updates and deletes" << endl ;
  ndbout << "   -use_ndb Use NDB API (otherwise use mysql client)" << endl ;
  ndbout << "   -pool_size Number of Ndb objects in pool" << endl ;
  theErrorData.printCmdLineArgs(ndbout);
  ndbout << endl <<"Returns:" << endl;
  ndbout << "\t 0 - Test passed" << endl;
  ndbout << "\t 1 - Test failed" << endl;
  ndbout << "\t 2 - Invalid arguments" << endl << endl;
}

// vim: set sw=2: