flex_bench_mysql.cpp

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

      bi.is_null = NULL;
    }//for
    
    for (Uint32 ca = 0; ca < 1; ca++){
      MYSQL_BIND& bi = bind_delete[ca];
      bi.buffer_type = MYSQL_TYPE_LONG;
      bi.buffer = (char*)&mysql_data[ca*tAttributeSize];
      bi.buffer_length = 0;
      bi.length = NULL;
      bi.is_null = NULL;
    }//for
    
    for (Uint32 i = 0; i < tNoOfTables; i++) {
      int pos = 0;
      pos += sprintf(buf+pos, "%s%s%s",
		     "INSERT INTO ",
		     tableName[i],
		     " VALUES(");
      pos += sprintf(buf+pos, "%s", "?");
      for (Uint32 j = 1; j < tNoOfAttributes; j++) {
	pos += sprintf(buf+pos, "%s", ",?");
      }
      pos += sprintf(buf+pos, "%s", ")");
      if (verbose)
	ndbout << buf << endl;
      prep_insert[i] = mysql_prepare(&mysql, buf, pos);
      if (prep_insert[i] == 0) {
	ndbout << "mysql_prepare: " << mysql_error(&mysql) << endl;
	return 0;
      }
      if (mysql_bind_param(prep_insert[i], bind_insert)) {
	ndbout << "mysql_bind_param: " << mysql_error(&mysql) << endl;
	return 0;
      }
    }
    
    for (Uint32 i = 0; i < tNoOfTables; i++) {
      int pos = 0;
      pos += sprintf(buf+pos, "%s%s%s",
		     "UPDATE ",
		     tableName[i],
		     " SET ");
      for (Uint32 j = 1; j < tNoOfAttributes; j++) {
	if (j != 1)
	  pos += sprintf(buf+pos, "%s", ",");
	pos += sprintf(buf+pos, "%s%s", attrName[j],"=?");
      }
      pos += sprintf(buf+pos, "%s%s%s", " WHERE ", attrName[0], "=?");
      
      if (verbose)
	ndbout << buf << endl;
      prep_update[i] = mysql_prepare(&mysql, buf, pos);
      if (prep_update[i] == 0) {
	ndbout << "mysql_prepare: " << mysql_error(&mysql) << endl;
	return 0;
      }
      if (mysql_bind_param(prep_update[i], bind_update)) {
	ndbout << "mysql_bind_param: " << mysql_error(&mysql) << endl;
	return 0;
      }
    }
    
    for (Uint32 i = 0; i < tNoOfTables; i++) {
      int pos = 0;
      pos += sprintf(buf+pos, "%s", "SELECT ");
      for (Uint32 j = 1; j < tNoOfAttributes; j++) {
	if (j != 1)
	  pos += sprintf(buf+pos, "%s", ",");
	pos += sprintf(buf+pos, "%s", attrName[j]);
      }
      pos += sprintf(buf+pos, "%s%s%s%s%s",
		     " FROM ",
		     tableName[i],
		     " WHERE ",
		     attrName[0],
		     "=?");
      if (verbose)
	ndbout << buf << endl;
      prep_read[i] = mysql_prepare(&mysql, buf, pos);
      if (prep_read[i] == 0) {
	ndbout << "mysql_prepare: " << mysql_error(&mysql) << endl;
	return 0;
      }
      if (mysql_bind_param(prep_read[i], bind_read)) {
	ndbout << "mysql_bind_param: " << mysql_error(&mysql) << endl;
	return 0;
      }
      if (mysql_bind_result(prep_read[i], &bind_read[1])) {
	ndbout << "mysql_bind_result: " << mysql_error(&mysql) << endl;
	return 0;
      }
    }
    
    for (Uint32 i = 0; i < tNoOfTables; i++) {
      int pos = 0;
      pos += sprintf(buf+pos, "%s%s%s%s%s",
		     "DELETE FROM ",
		     tableName[i],
		     " WHERE ",
		     attrName[0],
		     "=?");
      if (verbose)
	ndbout << buf << endl;
      prep_delete[i] = mysql_prepare(&mysql, buf, pos);
      if (prep_delete[i] == 0) {
	ndbout << "mysql_prepare: " << mysql_error(&mysql) << endl;
	return 0;
      }
      if (mysql_bind_param(prep_delete[i], bind_delete)) {
	ndbout << "mysql_bind_param: " << mysql_error(&mysql) << endl;
	return 0;
      }
    }
  }
#endif

  for (;;) {
    pThreadData->threadResult = tResult; // Report error to main thread, 
    // normally tResult is set to 0
    pThreadData->threadReady = 1;

    while (pThreadData->threadStart == stIdle){
      NdbSleep_MilliSleep(100);
    }//while

    // Check if signal to exit is received
    if (pThreadData->threadStart == stStop){
      pThreadData->threadReady = 1;
      // ndbout_c("Thread%d is stopping", threadNo);
      // In order to stop this thread, the main thread has signaled
      // stStop, break out of the for loop so that destructors
      // and the proper exit functions are called
      break;
    }//if

    tType = pThreadData->threadStart;
    tSaveType = tType;
    pThreadData->threadStart = stIdle;

    // Start transaction, type of transaction
    // is received in the array ThreadStart
    loopCountOps = tNoOfOperations;
    loopCountTables = tNoOfTables;
    loopCountAttributes = tNoOfAttributes;
    for (int count = 1; count < loopCountOps && tResult == 0;){

      if (use_ndb) {
        pNdb = get_ndb_object(ndb_id, "test", "def");
        if (pNdb == NULL) {
          ndbout << "Could not get Ndb object in thread" << threadNo;
          ndbout << endl;
          tResult = 1; //Indicate fatal error
          break;
        }
	pTrans = pNdb->startTransaction();
	if (pTrans == NULL) {
	  // This is a fatal error, abort program
	  ndbout << "Could not start transaction in thread" << threadNo;
	  ndbout << endl;
	  ndbout << pNdb->getNdbError() << endl;
	  tResult = 1; // Indicate fatal error
	  break; // Break out of for loop
	}
      }

      // Calculate the current operation offset in the reference array
      nRefLocalOpOffset = tAttributeSize*tNoOfAttributes*(count - 1) ;
      int* tmpAttrRefValue = attrRefValue + nRefLocalOpOffset;

      for (int countTables = 0;
	   countTables < loopCountTables && tResult == 0;
	   countTables++) {

	int nTableOffset = tAttributeSize *
	  loopCountAttributes *
	  countTables ;

	int* tmpAttrValue = attrValue + nTableOffset;

	if (use_ndb) {
	  pOps[countTables] = pTrans->getNdbOperation(tableName[countTables]); 
	  if (pOps[countTables] == NULL) {
	    // This is a fatal error, abort program
	    ndbout << "getNdbOperation: " << pTrans->getNdbError();
	    tResult = 2; // Indicate fatal error
	    break;
	  }//if

	  switch (tType) {
	  case stInsert:          // Insert case
	    if (theWriteFlag == 1 && theDirtyFlag == 1)
	    pOps[countTables]->dirtyWrite();
	    else if (theWriteFlag == 1)
	      pOps[countTables]->writeTuple();
	    else
	      pOps[countTables]->insertTuple();
	    break;
	  case stRead:            // Read Case
	    if (theSimpleFlag == 1)
	      pOps[countTables]->simpleRead();
	    else if (theDirtyFlag == 1)
	      pOps[countTables]->dirtyRead();
	    else
	      pOps[countTables]->readTuple();
	    break;
	  case stUpdate:          // Update Case
	    if (theWriteFlag == 1 && theDirtyFlag == 1)
	      pOps[countTables]->dirtyWrite();
	    else if (theWriteFlag == 1)
	      pOps[countTables]->writeTuple();
	    else if (theDirtyFlag == 1)
	      pOps[countTables]->dirtyUpdate();
	    else
	      pOps[countTables]->updateTuple();
	    break;
	  case stDelete:          // Delete Case
	    pOps[countTables]->deleteTuple();
	    break;
	  case stVerify:
	    pOps[countTables]->readTuple();
	    break;
	  case stVerifyDelete:
	    pOps[countTables]->readTuple();
	    break;
	  default:
	    assert(false);
	  }//switch
	  
	  if(useLongKeys){
	    // Loop the equal call so the complete key is send to the kernel.
	    for(Uint32 i = 0; i < tNoOfLongPK; i++) 
	      pOps[countTables]->equal(longKeyAttrName[i], 
				       (char *)longKeyAttrValue[count - 1][i],
				       tSizeOfLongPK*4); 
	  }
	  else 
	    pOps[countTables]->equal((char*)attrName[0], 
				     (char*)&tmpAttrRefValue[0]);

	  if (tType == stInsert) {
	    for (int ca = 1; ca < loopCountAttributes; ca++){
	      pOps[countTables]->setValue((char*)attrName[ca],
					  (char*)&tmpAttrRefValue[tAttributeSize*ca]);
	    }//for
	  } else if (tType == stUpdate) {
	    for (int ca = 1; ca < loopCountAttributes; ca++){
	      int* tmp = (int*)&tmpAttrRefValue[tAttributeSize*ca];
	      if (countTables == 0)
		(*tmp)++;
	      pOps[countTables]->setValue((char*)attrName[ca],(char*)tmp);
	    }//for
	  } else if (tType == stRead || stVerify == tType) {
	    for (int ca = 1; ca < loopCountAttributes; ca++) {
	      tTmp =
		pOps[countTables]->getValue((char*)attrName[ca], 
					    (char*)&tmpAttrValue[tAttributeSize*ca]);
	    }//for
	  } else if (stVerifyDelete == tType) {
	    if(useLongKeys){
	      tTmp = pOps[countTables]->getValue(longKeyAttrName[0], 
						 (char*)&tmpAttrValue[0]);
	    } else {
	      tTmp = pOps[countTables]->getValue((char*)attrName[0], 
						 (char*)&tmpAttrValue[0]);
	    }
	  }//if
	} else { // !use_ndb
#ifndef USE_MYSQL
	  assert(false);
#else
	  switch (tType)
	    {
	    case stInsert:
	      for (int ca = 0; ca < loopCountAttributes; ca++){
		mysql_data[ca] = tmpAttrRefValue[tAttributeSize*ca];
	      }//for
	      if (mysql_execute(prep_insert[countTables])) {
		ndbout << tableName[countTables];  
		ndbout << " mysql_execute: " << mysql_error(&mysql) << endl;
		tResult = 1 ;
	      }
	      break;
	    case stUpdate:          // Update Case
	      mysql_data[0] = tmpAttrRefValue[0];
	      for (int ca = 1; ca < loopCountAttributes; ca++){
		int* tmp = (int*)&tmpAttrRefValue[tAttributeSize*ca];
		if (countTables == 0)
		  (*tmp)++;
		mysql_data[ca] = *tmp;
	      }//for
	      if (mysql_execute(prep_update[countTables])) {
		ndbout << tableName[countTables];  
		ndbout << " mysql_execute: " << mysql_error(&mysql) << endl;
		tResult = 2 ;
	      }
	      break;
	    case stVerify:
	    case stRead:            // Read Case
	      mysql_data[0] = tmpAttrRefValue[0];
	      if (mysql_execute(prep_read[countTables])) {
		ndbout << tableName[countTables];  
		ndbout << " mysql_execute: " << mysql_error(&mysql) << endl;
		tResult = 3 ;
		break;
	      }
	      if (mysql_stmt_store_result(prep_read[countTables])) {
		ndbout << tableName[countTables];  
		ndbout << " mysql_stmt_store_result: "
		       << mysql_error(&mysql) << endl;
		tResult = 4 ;
		break;
	      }
	      {
		int rows= 0;
		int r;
		while ( (r= mysql_fetch(prep_read[countTables])) == 0 ){
		  rows++;
		}
		if ( r == 1 ) {
		  ndbout << tableName[countTables];  
		  ndbout << " mysql_fetch: " << mysql_error(&mysql) << endl;
		  tResult = 5 ;
		  break;
		}
		if ( rows != 1 ) {
		  ndbout << tableName[countTables];  
		  ndbout << " mysql_fetch: rows = " << rows << endl;
		  tResult = 6 ;
		  break;
		}
	      }
	      {
		for (int ca = 1; ca < loopCountAttributes; ca++) {
		  tmpAttrValue[tAttributeSize*ca] = mysql_data[ca];
		}
	      }
	      break;
	    case stDelete:          // Delete Case
	      mysql_data[0] = tmpAttrRefValue[0];
	      if (mysql_execute(prep_delete[countTables])) {
		ndbout << tableName[countTables];  
		ndbout << " mysql_execute: " << mysql_error(&mysql) << endl;
		tResult = 7 ;
		break;
	      }
	      break;
	    case stVerifyDelete:
	      {
		sprintf(buf, "%s%s%s",
			"SELECT COUNT(*) FROM ",tableName[countTables],";");
		if (mysql_query(&mysql, buf)) {
		  ndbout << buf << endl;
		  ndbout << "Error: " << mysql_error(&mysql) << endl;
		  tResult = 8 ;
		  break;
		}
		MYSQL_RES *res = mysql_store_result(&mysql);
		if ( res == NULL ) {
		  ndbout << "mysql_store_result: "
			 << mysql_error(&mysql) << endl
			 << "errno: " << mysql_errno(&mysql) << endl;
		  tResult = 9 ;
		  break;
		}
		int num_fields = mysql_num_fields(res);
		int num_rows   = mysql_num_rows(res);
		if ( num_rows != 1 || num_fields != 1 ) {
		  ndbout << tableName[countTables];  
		  ndbout << " mysql_store_result: num_rows = " << num_rows
			 << " num_fields = " << num_fields << endl;
		  tResult = 10 ;
		  break;
		}
		MYSQL_ROW row = mysql_fetch_row(res);
		if ( row == NULL ) {
		  ndbout << "mysql_fetch_row: "
			 << mysql_error(&mysql) << endl;
		  tResult = 11 ;
		  break;
		}
		if ( *(char*)row[0] != '0' ) {
		  ndbout << tableName[countTables];  
		  ndbout << " mysql_fetch_row: value = "
			 << (char*)(row[0]) << endl;
		  tResult = 12 ;
		  break;
		}
		mysql_free_result(res);
	      }
	      break;
	    default:
	      assert(false);
	    }
#endif
	}
      }//for Tables loop

      if (tResult != 0)
	break;

      if (use_ndb){
	check = pTrans->execute(Commit);
      } else {
#ifdef USE_MYSQL
	if (tNoOfTables > 1)
	  if (mysql_commit(&mysql)) {
	    ndbout << " mysql_commit: " << mysql_error(&mysql) << endl;
	    tResult = 13;
	  } else 
	    check = 0;
#endif
      }

      if (use_ndb) {
	// Decide what kind of error this is
	if ((tSpecialTrans == 1) &&
	    (check == -1)) {
// --------------------------------------------------------------------
// A special transaction have been executed, change to check = 0 in
// certain situations.
// --------------------------------------------------------------------
	  switch (tType) {
	  case stInsert:          // Insert case
	    if (630 == pTrans->getNdbError().code ) {
	      check = 0;
	      ndbout << "Insert with 4007 was successful" << endl;
	    }//if
	    break;
	  case stDelete:          // Delete Case
	    if (626 == pTrans->getNdbError().code ) {
	      check = 0;
	      ndbout << "Delete with 4007 was successful" << endl;
	    }//if
	    break;
	  default:
	    assert(false);
	  }//switch
	}//if