asyncgenerator.cpp

mysql-5.0.22.tar.gz源码包

{
  SessionElement  *se;
  
  /*----------------*/
  /* Init arguments */
  /*----------------*/
  se = getNextSession(&td->generator.activeSessions);
  if( se ) {
    strcpy(td->transactionData.number, se->subscriberNumber);
    td->transactionData.server_id = se->serverId;
    td->transactionData.sessionElement = 1;
  } else {
    getRandomSubscriberNumber(td->transactionData.number);
    getRandomServerId(&td->transactionData.server_id);
    td->transactionData.sessionElement = 0;
  }
  
  td->transactionData.server_bit = (1 << td->transactionData.server_id);

  /*-----------------*/
  /* Run transaction */
  /*-----------------*/
  td->runState = Running;
  td->generator.transactions[2].startLatency();
  start_T3(pNDB, td, async);
}

static 
void 
doTransaction_T4(Ndb * pNDB, ThreadData * td, int async)
{
   /*----------------*/
   /* Init arguments */
   /*----------------*/
  getRandomSubscriberNumber(td->transactionData.number);
  getRandomServerId(&td->transactionData.server_id);
  
  td->transactionData.server_bit = (1 << td->transactionData.server_id);
  td->transactionData.do_rollback = 
    getNextRandom(&td->generator.rollbackSequenceT4);

#if 0
  memset(td->transactionData.session_details, 
	 myRandom48(26)+'A', SESSION_DETAILS_LENGTH);
#endif
  td->transactionData.session_details[SESSION_DETAILS_LENGTH] = 0;
  
  /*-----------------*/
  /* Run transaction */
  /*-----------------*/
  td->runState = Running;
  td->generator.transactions[3].startLatency();
  start_T4(pNDB, td, async);
}

static 
void 
doTransaction_T5(Ndb * pNDB, ThreadData * td, int async)
{
  SessionElement * se;
  se = getNextSession(&td->generator.activeSessions);
  if( se ) {
    strcpy(td->transactionData.number, se->subscriberNumber);
    td->transactionData.server_id = se->serverId;
    td->transactionData.sessionElement = 1;
  }
  else {
    getRandomSubscriberNumber(td->transactionData.number);
    getRandomServerId(&td->transactionData.server_id);
    td->transactionData.sessionElement = 0;
  }
  
  td->transactionData.server_bit = (1 << td->transactionData.server_id);
  td->transactionData.do_rollback  
    = getNextRandom(&td->generator.rollbackSequenceT5);
  
  /*-----------------*/
  /* Run transaction */
  /*-----------------*/
  td->runState = Running;
  td->generator.transactions[4].startLatency();
  start_T5(pNDB, td, async);
}

void
complete_T1(ThreadData * data){
  data->generator.transactions[0].stopLatency();
  data->generator.transactions[0].count++;

  data->runState = Runnable;
  data->generator.totalTransactions++;
}

void 
complete_T2(ThreadData * data){
  data->generator.transactions[1].stopLatency();
  data->generator.transactions[1].count++;

  data->runState = Runnable;
  data->generator.totalTransactions++;
}

void 
complete_T3(ThreadData * data){

  data->generator.transactions[2].stopLatency();
  data->generator.transactions[2].count++;

  if(data->transactionData.branchExecuted)
    data->generator.transactions[2].branchExecuted++;

  data->runState = Runnable;
  data->generator.totalTransactions++;
}

void 
complete_T4(ThreadData * data){

  data->generator.transactions[3].stopLatency();
  data->generator.transactions[3].count++;

  if(data->transactionData.branchExecuted)
    data->generator.transactions[3].branchExecuted++;
  if(data->transactionData.do_rollback)
    data->generator.transactions[3].rollbackExecuted++;
  
  if(data->transactionData.branchExecuted &&
     !data->transactionData.do_rollback){
    insertSession(&data->generator.activeSessions, 
		  data->transactionData.number, 
		  data->transactionData.server_id);
  }

  data->runState = Runnable;
  data->generator.totalTransactions++;

}
void 
complete_T5(ThreadData * data){

  data->generator.transactions[4].stopLatency();
  data->generator.transactions[4].count++;

  if(data->transactionData.branchExecuted)
    data->generator.transactions[4].branchExecuted++;
  if(data->transactionData.do_rollback)
    data->generator.transactions[4].rollbackExecuted++;
  
  if(data->transactionData.sessionElement && 
     !data->transactionData.do_rollback){
    deleteSession(&data->generator.activeSessions);
  }
  
  data->runState = Runnable;
  data->generator.totalTransactions++;
}

/***************************************************************
****************************************************************
* P U B L I C   F U N C T I O N S   C O D E   S E C T I O N    *
****************************************************************
***************************************************************/
void 
asyncGenerator(ThreadData *data, 
	       int parallellism, 
	       int millisSendPoll,
	       int minEventSendPoll,
	       int forceSendPoll)
{
  ThreadData * startUp;
  
  GeneratorStatistics *st;
  double periodStop;
  double benchTimeStart;
  double benchTimeEnd;
  int i, j, done;

  myRandom48Init(data->randomSeed);
  
  for(i = 0; i<parallellism; i++){
    initGeneratorStatistics(&data[i].generator);
   }

  startUp = (ThreadData*)malloc(parallellism * sizeof(ThreadData));
  memcpy(startUp, data, (parallellism * sizeof(ThreadData)));
  
  /*----------------*/
  /* warm up period */
  /*----------------*/
  periodStop = userGetTime() + (double)data[0].warmUpSeconds;
  
  while(userGetTime() < periodStop){
    doOneTransaction(startUp, parallellism, 
		     millisSendPoll, minEventSendPoll, forceSendPoll);
  }
  
  ndbout_c("Waiting for startup to finish");

  /**
   * Wait for all transactions
   */
  done = 0;
  while(!done){
    done = 1;
    for(i = 0; i<parallellism; i++){
      if(startUp[i].runState != Runnable){
	done = 0;
	break;
      }
    }
    if(!done){
      startUp[0].pNDB->sendPollNdb();
    }
  }
  ndbout_c("Benchmark period starts");

  /*-------------------------*/
  /* normal benchmark period */
  /*-------------------------*/
  benchTimeStart = userGetTime();
  
  periodStop = benchTimeStart + (double)data[0].testSeconds;
  while(userGetTime() < periodStop)
    doOneTransaction(data, parallellism,
		     millisSendPoll, minEventSendPoll, forceSendPoll);  

  benchTimeEnd = userGetTime();
  
  ndbout_c("Benchmark period done");

  /**
   * Wait for all transactions
   */
  done = 0;
  while(!done){
    done = 1;
    for(i = 0; i<parallellism; i++){
      if(data[i].runState != Runnable){
	done = 0;
	break;
      }
    }
    if(!done){
      data[0].pNDB->sendPollNdb();
    }
  }

  /*------------------*/
  /* cool down period */
   /*------------------*/
  periodStop = userGetTime() + (double)data[0].coolDownSeconds;
  while(userGetTime() < periodStop){
    doOneTransaction(startUp, parallellism,
		     millisSendPoll, minEventSendPoll, forceSendPoll);
  }

  done = 0;
  while(!done){
    done = 1;
    for(i = 0; i<parallellism; i++){
      if(startUp[i].runState != Runnable){
	done = 0;
	break;
      }
    }
    if(!done){
      startUp[0].pNDB->sendPollNdb();
    }
  }


  /*---------------------------------------------------------*/
  /* add the times for all transaction for inner loop timing */
  /*---------------------------------------------------------*/
  for(j = 0; j<parallellism; j++){
    st = &data[j].generator;
    
    st->outerLoopTime = benchTimeEnd - benchTimeStart;
    st->outerTps      = getTps(st->totalTransactions, st->outerLoopTime);
  }
  /* ndbout_c("maxsize = %d\n",maxsize); */

  free(startUp);
}