开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > 数据库系统 > MySQL数据库 > 查看源码
asyncGenerator.cpp
资源名称：mysql-4.1.16-win-src.zip　[点击查看]
上传用户：romrleung
上传日期：2022-05-23
资源大小：18897k
文件大小：15k
源码类别：

MySQL数据库

开发平台：

Visual C++

asyncGenerator.cpp：源码内容
  1. /* Copyright (C) 2003 MySQL AB
  3.    This program is free software; you can redistribute it and/or modify
  4.    it under the terms of the GNU General Public License as published by
  5.    the Free Software Foundation; either version 2 of the License, or
  6.    (at your option) any later version.
  8.    This program is distributed in the hope that it will be useful,
  9.    but WITHOUT ANY WARRANTY; without even the implied warranty of
  10.    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11.    GNU General Public License for more details.
  13.    You should have received a copy of the GNU General Public License
  14.    along with this program; if not, write to the Free Software
  15.    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
  17. /***************************************************************
  18. * I N C L U D E D   F I L E S                                  *
  19. ***************************************************************/
  21. #include <ndb_global.h>
  23. #include "dbGenerator.h"
  24. #include <NdbApi.hpp>
  25. #include <NdbOut.hpp>
  26. #include <NdbSleep.h>
  28. /***************************************************************
  29. * L O C A L   C O N S T A N T S                                *
  30. ***************************************************************/
  32. /***************************************************************
  33. * L O C A L   D A T A   S T R U C T U R E S                    *
  34. ***************************************************************/
  36. /***************************************************************
  37. * L O C A L   F U N C T I O N S                                *
  38. ***************************************************************/
  40. static void getRandomSubscriberNumber(SubscriberNumber number);
  41. static void getRandomServerId(ServerId *serverId);
  42. static void getRandomChangedBy(ChangedBy changedBy);
  43. static void getRandomChangedTime(ChangedTime changedTime);
  45. static void clearTransaction(TransactionDefinition *trans);
  46. static void initGeneratorStatistics(GeneratorStatistics *gen);
  48. static void doOneTransaction(ThreadData * td, 
  49.      int parallellism,
  50.      int millisSendPoll,
  51.      int minEventSendPoll,
  52.      int forceSendPoll);
  53. static void doTransaction_T1(Ndb * pNDB, ThreadData * td, int async);
  54. static void doTransaction_T2(Ndb * pNDB, ThreadData * td, int async);
  55. static void doTransaction_T3(Ndb * pNDB, ThreadData * td, int async);
  56. static void doTransaction_T4(Ndb * pNDB, ThreadData * td, int async);
  57. static void doTransaction_T5(Ndb * pNDB, ThreadData * td, int async);
  59. /***************************************************************
  60. * L O C A L   D A T A                                          *
  61. ***************************************************************/
  63. static SequenceValues transactionDefinition[] = {
  64.    {25, 1},
  65.    {25, 2},
  66.    {20, 3},
  67.    {15, 4},
  68.    {15, 5},
  69.    {0,  0}
  70. };
  72. static SequenceValues rollbackDefinition[] = {
  73.    {98, 0},
  74.    {2 , 1},
  75.    {0,  0}
  76. };
  78. static int maxsize = 0;
  80. /***************************************************************
  81. * P U B L I C   D A T A                                        *
  82. ***************************************************************/
  84. /***************************************************************
  85. ****************************************************************
  86. * L O C A L   F U N C T I O N S   C O D E   S E C T I O N      *
  87. ****************************************************************
  88. ***************************************************************/
  90. static void getRandomSubscriberNumber(SubscriberNumber number)
  91. {
  92.    uint32 tmp;
  93.    char sbuf[SUBSCRIBER_NUMBER_LENGTH + 1];
  94.    tmp = myRandom48(NO_OF_SUBSCRIBERS);
  95.    sprintf(sbuf, "%.*d", SUBSCRIBER_NUMBER_LENGTH, tmp);
  96.    memcpy(number, sbuf, SUBSCRIBER_NUMBER_LENGTH);
  97. }
  99. static void getRandomServerId(ServerId *serverId)
  100. {
  101.    *serverId = myRandom48(NO_OF_SERVERS);
  102. }
  104. static void getRandomChangedBy(ChangedBy changedBy)
  105. {
  106.    memset(changedBy, myRandom48(26)+'A', CHANGED_BY_LENGTH);
  107.    changedBy[CHANGED_BY_LENGTH] = 0;
  108. }
  110. static void getRandomChangedTime(ChangedTime changedTime)
  111. {
  112.    memset(changedTime, myRandom48(26)+'A', CHANGED_TIME_LENGTH);
  113.    changedTime[CHANGED_TIME_LENGTH] = 0;
  114. }
  116. static void clearTransaction(TransactionDefinition *trans)
  117. {
  118.   trans->count            = 0;
  119.   trans->branchExecuted   = 0;
  120.   trans->rollbackExecuted = 0;
  121.   trans->latencyCounter   = myRandom48(127);
  122.   trans->latency.reset();
  123. }
  125. static int listFull(SessionList *list)
  126. {
  127.    return(list->numberInList == SESSION_LIST_LENGTH);
  128. }
  130. static int listEmpty(SessionList *list)
  131. {
  132.    return(list->numberInList == 0);
  133. }
  135. static void insertSession(SessionList     *list, 
  136.                           SubscriberNumber number,
  137.                           ServerId         serverId)
  138. {
  139.    SessionElement *e;
  140.    if( listFull(list) ) return;
  142.    e = &list->list[list->writeIndex];
  144.    strcpy(e->subscriberNumber, number);
  145.    e->serverId = serverId;
  147.    list->writeIndex = (list->writeIndex + 1) % SESSION_LIST_LENGTH;
  148.    list->numberInList++;
  150.    if( list->numberInList > maxsize )
  151.      maxsize = list->numberInList;
  152. }
  154. static SessionElement *getNextSession(SessionList *list)
  155. {
  156.    if( listEmpty(list) ) return(0);
  158.    return(&list->list[list->readIndex]);
  159. }
  161. static void deleteSession(SessionList *list)
  162. {
  163.    if( listEmpty(list) ) return;
  165.    list->readIndex = (list->readIndex + 1) % SESSION_LIST_LENGTH;
  166.    list->numberInList--;
  167. }
  169. static void initGeneratorStatistics(GeneratorStatistics *gen)
  170. {
  171.    int i;
  173.    if( initSequence(&gen->transactionSequence,
  174.                     transactionDefinition) != 0 ) {
  175.       ndbout_c("could not set the transaction types");
  176.       exit(0);
  177.    }
  179.    if( initSequence(&gen->rollbackSequenceT4,
  180.                     rollbackDefinition) != 0 ) {
  181.       ndbout_c("could not set the rollback sequence");
  182.       exit(0);
  183.    }
  185.    if( initSequence(&gen->rollbackSequenceT5,
  186.                     rollbackDefinition) != 0 ) {
  187.       ndbout_c("could not set the rollback sequence");
  188.       exit(0);
  189.    }
  191.    for(i = 0; i < NUM_TRANSACTION_TYPES; i++ )
  192.       clearTransaction(&gen->transactions[i]);
  194.    gen->totalTransactions = 0;
  196.    gen->activeSessions.numberInList = 0;
  197.    gen->activeSessions.readIndex    = 0;
  198.    gen->activeSessions.writeIndex   = 0;
  199. }
  202. static 
  203. void 
  204. doOneTransaction(ThreadData * td, int p, int millis, int minEvents, int force)
  205. {
  206.   int i;
  207.   unsigned int transactionType;
  208.   int async = 1;
  209.   if (p == 1) {
  210.     async = 0;
  211.   }//if
  212.   for(i = 0; i<p; i++){
  213.     if(td[i].runState == Runnable){
  214.       transactionType = getNextRandom(&td[i].generator.transactionSequence);
  216.       switch(transactionType) {
  217.       case 1:
  218. doTransaction_T1(td[i].pNDB, &td[i], async);
  219. break;
  220.       case 2:
  221. doTransaction_T2(td[i].pNDB, &td[i], async);
  222. break;
  223.       case 3:
  224. doTransaction_T3(td[i].pNDB, &td[i], async);
  225. break;
  226.       case 4:
  227. doTransaction_T4(td[i].pNDB, &td[i], async);
  228. break;
  229.       case 5:
  230. doTransaction_T5(td[i].pNDB, &td[i], async);
  231. break;
  232.       default:
  233. ndbout_c("Unknown transaction type: %d", transactionType);
  234.       }
  235.     }
  236.   }
  237.   if (async == 1) {
  238.     td[0].pNDB->sendPollNdb(millis, minEvents, force);
  239.   }//if
  240. }  
  242. static 
  243. void 
  244. doTransaction_T1(Ndb * pNDB, ThreadData * td, int async)
  245. {
  246.   /*----------------*/
  247.   /* Init arguments */
  248.   /*----------------*/
  249.   getRandomSubscriberNumber(td->transactionData.number);
  250.   getRandomChangedBy(td->transactionData.changed_by);
  251.   BaseString::snprintf(td->transactionData.changed_time,
  252.    sizeof(td->transactionData.changed_time),
  253.    "%ld - %d", td->changedTime++, myRandom48(65536*1024));
  254.   //getRandomChangedTime(td->transactionData.changed_time);
  255.   td->transactionData.location = td->transactionData.changed_by[0];
  256.   
  257.   /*-----------------*/
  258.   /* Run transaction */
  259.   /*-----------------*/
  260.   td->runState = Running;
  261.   td->generator.transactions[0].startLatency();
  263.   start_T1(pNDB, td, async);
  264. }
  266. static
  267. void 
  268. doTransaction_T2(Ndb * pNDB, ThreadData * td, int async)
  269. {
  270.   /*----------------*/
  271.   /* Init arguments */
  272.   /*----------------*/
  273.   getRandomSubscriberNumber(td->transactionData.number);
  275.   /*-----------------*/
  276.   /* Run transaction */
  277.   /*-----------------*/
  278.   td->runState = Running;
  279.   td->generator.transactions[1].startLatency();
  281.   start_T2(pNDB, td, async);
  282. }
  284. static
  285. void 
  286. doTransaction_T3(Ndb * pNDB, ThreadData * td, int async)
  287. {
  288.   SessionElement  *se;
  289.   
  290.   /*----------------*/
  291.   /* Init arguments */
  292.   /*----------------*/
  293.   se = getNextSession(&td->generator.activeSessions);
  294.   if( se ) {
  295.     strcpy(td->transactionData.number, se->subscriberNumber);
  296.     td->transactionData.server_id = se->serverId;
  297.     td->transactionData.sessionElement = 1;
  298.   } else {
  299.     getRandomSubscriberNumber(td->transactionData.number);
  300.     getRandomServerId(&td->transactionData.server_id);
  301.     td->transactionData.sessionElement = 0;
  302.   }
  303.   
  304.   td->transactionData.server_bit = (1 << td->transactionData.server_id);
  306.   /*-----------------*/
  307.   /* Run transaction */
  308.   /*-----------------*/
  309.   td->runState = Running;
  310.   td->generator.transactions[2].startLatency();
  311.   start_T3(pNDB, td, async);
  312. }
  314. static 
  315. void 
  316. doTransaction_T4(Ndb * pNDB, ThreadData * td, int async)
  317. {
  318.    /*----------------*/
  319.    /* Init arguments */
  320.    /*----------------*/
  321.   getRandomSubscriberNumber(td->transactionData.number);
  322.   getRandomServerId(&td->transactionData.server_id);
  323.   
  324.   td->transactionData.server_bit = (1 << td->transactionData.server_id);
  325.   td->transactionData.do_rollback = 
  326.     getNextRandom(&td->generator.rollbackSequenceT4);
  328. #if 0
  329.   memset(td->transactionData.session_details, 
  330.  myRandom48(26)+'A', SESSION_DETAILS_LENGTH);
  331. #endif
  332.   td->transactionData.session_details[SESSION_DETAILS_LENGTH] = 0;
  333.   
  334.   /*-----------------*/
  335.   /* Run transaction */
  336.   /*-----------------*/
  337.   td->runState = Running;
  338.   td->generator.transactions[3].startLatency();
  339.   start_T4(pNDB, td, async);
  340. }
  342. static 
  343. void 
  344. doTransaction_T5(Ndb * pNDB, ThreadData * td, int async)
  345. {
  346.   SessionElement * se;
  347.   se = getNextSession(&td->generator.activeSessions);
  348.   if( se ) {
  349.     strcpy(td->transactionData.number, se->subscriberNumber);
  350.     td->transactionData.server_id = se->serverId;
  351.     td->transactionData.sessionElement = 1;
  352.   }
  353.   else {
  354.     getRandomSubscriberNumber(td->transactionData.number);
  355.     getRandomServerId(&td->transactionData.server_id);
  356.     td->transactionData.sessionElement = 0;
  357.   }
  358.   
  359.   td->transactionData.server_bit = (1 << td->transactionData.server_id);
  360.   td->transactionData.do_rollback  
  361.     = getNextRandom(&td->generator.rollbackSequenceT5);
  362.   
  363.   /*-----------------*/
  364.   /* Run transaction */
  365.   /*-----------------*/
  366.   td->runState = Running;
  367.   td->generator.transactions[4].startLatency();
  368.   start_T5(pNDB, td, async);
  369. }
  371. void
  372. complete_T1(ThreadData * data){
  373.   data->generator.transactions[0].stopLatency();
  374.   data->generator.transactions[0].count++;
  376.   data->runState = Runnable;
  377.   data->generator.totalTransactions++;
  378. }
  380. void 
  381. complete_T2(ThreadData * data){
  382.   data->generator.transactions[1].stopLatency();
  383.   data->generator.transactions[1].count++;
  385.   data->runState = Runnable;
  386.   data->generator.totalTransactions++;
  387. }
  389. void 
  390. complete_T3(ThreadData * data){
  392.   data->generator.transactions[2].stopLatency();
  393.   data->generator.transactions[2].count++;
  395.   if(data->transactionData.branchExecuted)
  396.     data->generator.transactions[2].branchExecuted++;
  398.   data->runState = Runnable;
  399.   data->generator.totalTransactions++;
  400. }
  402. void 
  403. complete_T4(ThreadData * data){
  405.   data->generator.transactions[3].stopLatency();
  406.   data->generator.transactions[3].count++;
  408.   if(data->transactionData.branchExecuted)
  409.     data->generator.transactions[3].branchExecuted++;
  410.   if(data->transactionData.do_rollback)
  411.     data->generator.transactions[3].rollbackExecuted++;
  412.   
  413.   if(data->transactionData.branchExecuted &&
  414.      !data->transactionData.do_rollback){
  415.     insertSession(&data->generator.activeSessions, 
  416.   data->transactionData.number, 
  417.   data->transactionData.server_id);
  418.   }
  420.   data->runState = Runnable;
  421.   data->generator.totalTransactions++;
  423. }
  424. void 
  425. complete_T5(ThreadData * data){
  427.   data->generator.transactions[4].stopLatency();
  428.   data->generator.transactions[4].count++;
  430.   if(data->transactionData.branchExecuted)
  431.     data->generator.transactions[4].branchExecuted++;
  432.   if(data->transactionData.do_rollback)
  433.     data->generator.transactions[4].rollbackExecuted++;
  434.   
  435.   if(data->transactionData.sessionElement && 
  436.      !data->transactionData.do_rollback){
  437.     deleteSession(&data->generator.activeSessions);
  438.   }
  439.   
  440.   data->runState = Runnable;
  441.   data->generator.totalTransactions++;
  442. }
  444. /***************************************************************
  445. ****************************************************************
  446. * 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    *
  447. ****************************************************************
  448. ***************************************************************/
  449. void 
  450. asyncGenerator(ThreadData *data, 
  451.        int parallellism, 
  452.        int millisSendPoll,
  453.        int minEventSendPoll,
  454.        int forceSendPoll)
  455. {
  456.   ThreadData * startUp;
  457.   
  458.   GeneratorStatistics *st;
  459.   double periodStop;
  460.   double benchTimeStart;
  461.   double benchTimeEnd;
  462.   int i, j, done;
  464.   myRandom48Init(data->randomSeed);
  465.   
  466.   for(i = 0; i<parallellism; i++){
  467.     initGeneratorStatistics(&data[i].generator);
  468.    }
  470.   startUp = (ThreadData*)malloc(parallellism * sizeof(ThreadData));
  471.   memcpy(startUp, data, (parallellism * sizeof(ThreadData)));
  472.   
  473.   /*----------------*/
  474.   /* warm up period */
  475.   /*----------------*/
  476.   periodStop = userGetTime() + (double)data[0].warmUpSeconds;
  477.   
  478.   while(userGetTime() < periodStop){
  479.     doOneTransaction(startUp, parallellism, 
  480.      millisSendPoll, minEventSendPoll, forceSendPoll);
  481.   }
  482.   
  483.   ndbout_c("Waiting for startup to finish");
  485.   /**
  486.    * Wait for all transactions
  487.    */
  488.   done = 0;
  489.   while(!done){
  490.     done = 1;
  491.     for(i = 0; i<parallellism; i++){
  492.       if(startUp[i].runState != Runnable){
  493. done = 0;
  494. break;
  495.       }
  496.     }
  497.     if(!done){
  498.       startUp[0].pNDB->sendPollNdb();
  499.     }
  500.   }
  501.   ndbout_c("Benchmark period starts");
  503.   /*-------------------------*/
  504.   /* normal benchmark period */
  505.   /*-------------------------*/
  506.   benchTimeStart = userGetTime();
  507.   
  508.   periodStop = benchTimeStart + (double)data[0].testSeconds;
  509.   while(userGetTime() < periodStop)
  510.     doOneTransaction(data, parallellism,
  511.      millisSendPoll, minEventSendPoll, forceSendPoll);  
  513.   benchTimeEnd = userGetTime();
  514.   
  515.   ndbout_c("Benchmark period done");
  517.   /**
  518.    * Wait for all transactions
  519.    */
  520.   done = 0;
  521.   while(!done){
  522.     done = 1;
  523.     for(i = 0; i<parallellism; i++){
  524.       if(data[i].runState != Runnable){
  525. done = 0;
  526. break;
  527.       }
  528.     }
  529.     if(!done){
  530.       data[0].pNDB->sendPollNdb();
  531.     }
  532.   }
  534.   /*------------------*/
  535.   /* cool down period */
  536.    /*------------------*/
  537.   periodStop = userGetTime() + (double)data[0].coolDownSeconds;
  538.   while(userGetTime() < periodStop){
  539.     doOneTransaction(startUp, parallellism,
  540.      millisSendPoll, minEventSendPoll, forceSendPoll);
  541.   }
  543.   done = 0;
  544.   while(!done){
  545.     done = 1;
  546.     for(i = 0; i<parallellism; i++){
  547.       if(startUp[i].runState != Runnable){
  548. done = 0;
  549. break;
  550.       }
  551.     }
  552.     if(!done){
  553.       startUp[0].pNDB->sendPollNdb();
  554.     }
  555.   }
  558.   /*---------------------------------------------------------*/
  559.   /* add the times for all transaction for inner loop timing */
  560.   /*---------------------------------------------------------*/
  561.   for(j = 0; j<parallellism; j++){
  562.     st = &data[j].generator;
  563.     
  564.     st->outerLoopTime = benchTimeEnd - benchTimeStart;
  565.     st->outerTps      = getTps(st->totalTransactions, st->outerLoopTime);
  566.   }
  567.   /* ndbout_c("maxsize = %dn",maxsize); */
  569.   free(startUp);
  570. }