flexhammer.cpp

mysql-5.0.22.tar.gz源码包

	    } else {
	      check = pMyOperation[count_tables]->insertTuple();
	    } // if else
	    break;
	  case otRead:			// Read Case
	    if (theSimpleFlag == 1) {
	      check = pMyOperation[count_tables]->simpleRead();
	    } else if (theDirtyFlag == 1) {
	      check = pMyOperation[count_tables]->dirtyRead();
	    } else {
	      check = pMyOperation[count_tables]->readTuple();
	    } // if else
	    break;
	  case otUpdate:			// Update Case
	    if (theWriteFlag == 1 && theDirtyFlag == 1) {
	      check = pMyOperation[count_tables]->dirtyWrite();
	    } else if (theWriteFlag == 1) {
	      check = pMyOperation[count_tables]->writeTuple();
	    } else if (theDirtyFlag == 1) {
	      check = pMyOperation[count_tables]->dirtyUpdate();
	    } else {
	      check = pMyOperation[count_tables]->updateTuple();
	    } // if else
	    break;
	  case otDelete:			// Delete Case
	    check = pMyOperation[count_tables]->deleteTuple();
	    break;
	  default:
	    assert(false);
	    break;
	  } // switch
	  if (check == -1) {
	    // Fatal error
	    tThreadResult = 3;
	    // break out of inner for count_tables loop
	    break;
	  } // if

	  check = pMyOperation[count_tables]->equal( (char*)attrName[0], 
						    (char*)&pkValue );

	  if (check == -1) {
	    // Fatal error
	    tThreadResult = 4;
	    ndbout << "pMyOperation equal failed" << endl;
	    // break out of inner for count_tables loop
	    break;
	  } // if
	  
	  check = -1;
	  tTmp = NULL;
	  switch (tMyOpType) {
	  case otInsert:			// Insert case
	  case otUpdate:			// Update Case
	    for (count_attributes = 1; count_attributes < loop_count_attributes;
		 count_attributes++) {
	      check = 
		pMyOperation[count_tables]->setValue((char*)attrName[count_attributes], (char*)&attrValue[0]);
	    } // for
	    break;
	  case otRead:			// Read Case
	    for (count_attributes = 1; count_attributes < loop_count_attributes; 
		 count_attributes++) {
	      tTmp = pMyOperation[count_tables]->
		getValue( (char*)attrName[count_attributes], 
			  (char*)&readValue[count_attributes][0] );
	    } // for
	    break;
	  case otDelete:			// Delete Case
	    break;
	  default:
	    assert(false);
	    break;
	  } // switch
	  if (check == -1 && tTmp == NULL && tMyOpType != otDelete) {
	    // Fatal error
	    tThreadResult = 5;
	    break;
	  } // if
	} // for count_tables
	    
	// Only execute if everything is OK
	if (tThreadResult != 0) {
	  // Close transaction (below)
	  // and continue with next count_round
	  count_round++;
	  tNoOfAttempts = 0;
	} // if
	else {
	  check = pMyTransaction->execute(Commit);
	  if (check == -1 ) {
	    const NdbError & err = pMyTransaction->getNdbError();

	// Add complete error handling here

              int retCode = flexHammerErrorData->handleErrorCommon(pMyTransaction->getNdbError());
              if (retCode == 1) {
		//if (strcmp(pMyTransaction->getNdbError().message, "Tuple did not exist") != 0 && strcmp(pMyTransaction->getNdbError().message,"Tuple already existed when attempting to insert") != 0) ndbout_c("execute: %s", pMyTransaction->getNdbError().message);

		if (pMyTransaction->getNdbError().code != 626 && pMyTransaction->getNdbError().code != 630){
     ndbout_c("Error code = %d", pMyTransaction->getNdbError().code);
     ndbout_c("execute: %s", pMyTransaction->getNdbError().message);}

              } else if (retCode == 2) {
                ndbout << "4115 should not happen in flexHammer" << endl;
              } 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.
// --------------------------------------------------------------------
		/* What can we do here? */
		ndbout_c("execute: %s", pMyTransaction->getNdbError().message);
                 }//if(retCode == 3)
	// End of adding complete error handling

	    switch( err.classification) {
	    case NdbError::ConstraintViolation:	// Tuple already existed
	      count_round++;
	      tNoOfAttempts = 0;
	      break;
	    case NdbError::TimeoutExpired:
	    case NdbError::NodeRecoveryError:
	    case NdbError::TemporaryResourceError:
	    case NdbError::OverloadError:
	      if (tNoOfAttempts <= MaxNoOfAttemptsC) {
		// Retry
		tNoOfAttempts++;
	      } else {
		// Too many retries, continue with next
		count_round++;
		tNoOfAttempts = 0;
	      } // else if
	      break;
	      // Fatal, just continue
	    default:
	      count_round++;
	      tNoOfAttempts = 0;
	      break;
	    } // switch
	  } // if
	  else {
	    // Execute commit was OK
	    // This is verifying read values
	    //switch (tMyOpType) {
	    //case otRead:  // Read case
	    //for (j = 0; j < tNoOfAttributes; j++) {
	    //for(i = 1; i < tAttributeSize; i++) {
	    //if ( readValue[j][i] != attrValue[i]) {
	    //ndbout << "pkValue = " << pkValue << endl;
	    //ndbout << "readValue != attrValue" << endl;
	    //ndbout << readValue[j][i] << " != " << attrValue[i] << endl;
	    //} // if
	    //  } // for
	    //} // for
	    //break;
	    //} // switch
	    count_round++;
	    tNoOfAttempts = 0;
	  } // else if
	} // else if
	pMyNdb->closeTransaction(pMyTransaction);
      } // for count_round
    } // for count
  } // for (;;)

  // Clean up
  delete pMyNdb; 
  pMyNdb = NULL;

  flexHammerErrorData->resetErrorCounters();

  return  NULL; // thread exits
  
} // flexHammerThread


int
readArguments (int argc, const char** argv)
{
  int i = 1;
  
  tNoOfThreads = 5;		// Default Value
  tNoOfOperations = 500;	// Default Value
  tNoOfRecords = 1;             // Default Value
  tNoOfLoops = 1;	        // Default Value
  tNoOfAttributes = 25;		// Default Value
  tNoOfTables = 1;		// Default Value
  tNoOfBackups = 0;		// Default Value
  tAttributeSize = 1;		// Default Value
  theTableCreateFlag = 0;
  
  while (argc > 1) {
    if (strcmp(argv[i], "-t") == 0) {
      tNoOfThreads = atoi(argv[i+1]);
      if ((tNoOfThreads < 1) || (tNoOfThreads > MAXTHREADS))
	return(1);
    }
    else if (strcmp(argv[i], "-o") == 0) {
      tNoOfOperations = atoi(argv[i+1]);
      if (tNoOfOperations < 1)
	return(1);
    }
    else if (strcmp(argv[i], "-r") == 0) {
      tNoOfRecords = atoi(argv[i+1]);
      if (tNoOfRecords < 1)
	return(1);
    }
    else if (strcmp(argv[i], "-a") == 0) {
      tNoOfAttributes = atoi(argv[i+1]);
      if ((tNoOfAttributes < 2) || (tNoOfAttributes > MAXATTR))
	return(1);
    }
    else if (strcmp(argv[i], "-c") == 0) {
      tNoOfTables = atoi(argv[i+1]);
      if ((tNoOfTables < 1) || (tNoOfTables > MAXTABLES))
	return(1);
    }
    else if (strcmp(argv[i], "-l") == 0) {
      tNoOfLoops = atoi(argv[i+1]);
      if ((tNoOfLoops < 0) || (tNoOfLoops > 100000))
	return(1);
    }
    else if (strcmp(argv[i], "-s") == 0) {
      tAttributeSize = atoi(argv[i+1]);
      if ((tAttributeSize < 1) || (tAttributeSize > MAXATTRSIZE))
	return(1);
    }
    else if (strcmp(argv[i], "-sleep") == 0) {
      tSleepTime = atoi(argv[i+1]);
      if ((tSleepTime < 1) || (tSleepTime > 3600))
	exit(-1);
    }
    else if (strcmp(argv[i], "-simple") == 0) {
      theSimpleFlag = 1;
      argc++;
      i--;
    }
    else if (strcmp(argv[i], "-write") == 0) {
      theWriteFlag = 1;
      argc++;
      i--;
    }
    else if (strcmp(argv[i], "-dirty") == 0) {
      theDirtyFlag = 1;
      argc++;
      i--;
    }
    else if (strcmp(argv[i], "-no_table_create") == 0) {
      theTableCreateFlag = 1;
      argc++;
      i--;
    }
    else if (strcmp(argv[i], "-stdtables") == 0) {
      theStandardTableNameFlag = 1;
      argc++;
      i--;
    } // if
    else {
      return(1);
    }
    
    argc -= 2;
    i = i + 2;
  } // while
  
  ndbout << endl << "FLEXHAMMER - Starting normal mode" << endl;
  ndbout << "Hammer ndb with read, insert, update and delete transactions"<< endl << endl;
  
  ndbout << "  " << tNoOfThreads << " thread(s) " << endl;
  ndbout << "  " << tNoOfLoops << " iterations " << endl;
  ndbout << "  " << tNoOfTables << " table(s) and " << 1 << " operation(s) per transaction " << endl;
  ndbout << "  " << tNoOfRecords << " records to hammer(limit this with the -r option)" << 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 << endl;
  return 0;
} // readArguments


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

static int
createTables(Ndb* pMyNdb)
{
  int i = 0;
  int j = 0;
  int check = 0;
  NdbSchemaCon *MySchemaTransaction = NULL;
  NdbSchemaOp *MySchemaOp = NULL;

  //	Create Table and Attributes. 	                          
  if (theTableCreateFlag == 0) {

    for (i = 0; i < tNoOfTables; i++) {

      ndbout << "Creating " << tableName[i] << "...";
      // Check if table exists already
      const void * p = pMyNdb->getDictionary()->getTable(tableName[i]);
      if (p != 0) {
	ndbout << " already exists." << endl;
	// Continue with next table at once
	continue;
      } // if
      ndbout << endl;
      
      MySchemaTransaction = NdbSchemaCon::startSchemaTrans(pMyNdb);
      if (MySchemaTransaction == NULL) {
	return(-1);
      } // if
      
      MySchemaOp = MySchemaTransaction->getNdbSchemaOp();	
      if (MySchemaOp == NULL) {
	// Clean up opened schema transaction
	NdbSchemaCon::closeSchemaTrans(MySchemaTransaction);
	return(-1);
      } // if
      
      // Create tables, rest of parameters are default right now
#if defined NDB_OSE || defined NDB_SOFTOSE
      check = MySchemaOp->createTable(tableName[i],
				      8,		// Table Size
				      TupleKey,	// Key Type
				      40,		// Nr of Pages
				      All, 
				      6,
				      78,
				      80,
				      1,
				      false);

#else
      check = MySchemaOp->createTable(tableName[i],
				      8,		// Table Size
				      TupleKey,	// Key Type
				      40);		// Nr of Pages
#endif
      if (check == -1) { 
	// Clean up opened schema transaction
	NdbSchemaCon::closeSchemaTrans(MySchemaTransaction);
	return(-1);
      } // if
      
      // Primary key
      //ndbout << "  pk " << (char*)&attrName[0] << "..." << endl;
      check = MySchemaOp->createAttribute( (char*)attrName[0], TupleKey, 32,
					   1, UnSigned, MMBased,
					   NotNullAttribute );
      if (check == -1) { 
	// Clean up opened schema transaction
	NdbSchemaCon::closeSchemaTrans(MySchemaTransaction);
	return(-1);
      } // if

      // Rest of attributes
      for (j = 1; j < tNoOfAttributes ; j++) {
	//ndbout << "    " << (char*)attrName[j] << "..." << endl;
	check = MySchemaOp->createAttribute( (char*)attrName[j], NoKey, 32, 
					     tAttributeSize, UnSigned, MMBased, 
					     NotNullAttribute );
	if (check == -1) {
	  // Clean up opened schema transaction
	  NdbSchemaCon::closeSchemaTrans(MySchemaTransaction);
	  return(-1);
	} // if
      } // for
  
      // Execute creation
      check = MySchemaTransaction->execute();
      if (check == -1) {
	// Clean up opened schema transaction
	NdbSchemaCon::closeSchemaTrans(MySchemaTransaction);
	return(-1);
      } // if
      
      NdbSchemaCon::closeSchemaTrans(MySchemaTransaction);
    } // for
  } // if

  return(0);

} // createTables 


static int setAttrNames()
{
  int i = 0;
  int retVal = 0;

  for (i = 0; i < MAXATTR ; i++) {
    retVal = BaseString::snprintf(attrName[i], MAXSTRLEN, "COL%d", i);
    if (retVal < 0) {
      // Error in conversion
      return(-1);
    } // if
  } // for

  return (0);
} // setAttrNames

static int setTableNames()
{
  // Note! Uses only uppercase letters in table name's
  // so that we can look at the tables wits SQL
  int i = 0;
  int retVal = 0;

  for (i = 0; i < MAXTABLES ; i++) {
    if (theStandardTableNameFlag == 0) {
      retVal = BaseString::snprintf(tableName[i], MAXSTRLEN, "TAB%d_%d", i, 
		 NdbTick_CurrentMillisecond()/1000);
    } // if 
    else {
      retVal = BaseString::snprintf(tableName[i], MAXSTRLEN, "TAB%d", i);
    } // else
    if (retVal < 0) {
      // Error in conversion
      return(-1);
    } // if
  } // for

  return(0);
} // setTableNames

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

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

  return(0);
}