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

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

MySQL数据库

开发平台：

Visual C++

flex_bench_mysql.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 */
  18. /* ***************************************************
  19. FLEXBENCH
  20. Perform benchmark of insert, update and delete transactions
  22. Arguments:
  23.     -t Number of threads to start, default 1
  24.     -o Number of operations per loop, default 500
  25.     -l Number of loops to run, default 1, 0=infinite
  26.     -a Number of attributes, default 25
  27.     -c Number of tables, default 1
  28.     -s Size of each attribute, default 1 (Primary Key is always of size 1,
  29.     independent of this value)
  30.     -lkn Number of long primary keys, default 1
  31.     -lks Size of each long primary key, default 1
  32.     -simple Use simple read to read from database
  33.     -dirty Use dirty read to read from database
  34.     -write Use writeTuple in insert and update
  35.     -stdtables Use standard table names
  36.     -no_table_create Don't create tables in db
  37.     -sleep Sleep a number of seconds before running the test, this 
  38.     can be used so that another flexBench have time to create tables
  39.     -temp Use tables without logging
  40.     -verify Verify inserts, updates and deletes
  41.     -use_ndb Use NDB API, otherwise use mysql client
  42. #ifdef CEBIT_STAT
  43.     -statserv host:port  statistics server to report to
  44.     -statfreq ops        report every ops operations (default 100)
  45. #endif
  46.     Returns:
  47.     0 - Test passed
  48.     1 - Test failed
  49.     2 - Invalid arguments
  51. * *************************************************** */
  53. #define USE_MYSQL
  54. #ifdef USE_MYSQL
  55. #include <mysql.h>
  56. #endif
  58. #include "NdbApi.hpp"
  60. #include <NdbMain.h>
  61. #include <NdbOut.hpp>
  62. #include <NdbSleep.h>
  63. #include <NdbTick.h>
  64. #include <NdbTimer.hpp>
  65. #include <NdbThread.h>
  66. #include <NdbAutoPtr.hpp>
  68. #include <NdbTest.hpp>
  70. #define MAXSTRLEN 16 
  71. #define MAXATTR 64
  72. #define MAXTABLES 128
  73. #define MAXATTRSIZE 1000
  74. #define MAXNOLONGKEY 16 // Max number of long keys.
  75. #define MAXLONGKEYTOTALSIZE 1023 // words = 4092 bytes
  77. extern "C" { static void* flexBenchThread(void*); }
  78. static int readArguments(int argc, const char** argv);
  79. #ifdef USE_MYSQL
  80. static int createTables(MYSQL*);
  81. static int dropTables(MYSQL*);
  82. #endif
  83. static int createTables(Ndb*);
  84. static void sleepBeforeStartingTest(int seconds);
  85. static void input_error();
  87. enum StartType { 
  88.   stIdle,
  89.   stInsert,
  90.   stVerify,
  91.   stRead,
  92.   stUpdate,
  93.   stDelete,
  94.   stTryDelete,
  95.   stVerifyDelete,
  96.   stStop 
  97. };
  99. struct ThreadData
  100. {
  101.   int threadNo;
  102.   NdbThread* threadLife;
  103.   int threadReady;  
  104.   StartType threadStart;
  105.   int threadResult;
  106. };
  108. static int                  tNodeId = 0 ;
  109. static char                 tableName[MAXTABLES][MAXSTRLEN+1];
  110. static char                 attrName[MAXATTR][MAXSTRLEN+1];
  111. static char**               longKeyAttrName;
  113. // Program Parameters
  114. static int                  tNoOfLoops = 1;
  115. static int                  tAttributeSize = 1;
  116. static unsigned int         tNoOfThreads = 1;
  117. static unsigned int         tNoOfTables = 1;
  118. static unsigned int         tNoOfAttributes = 25;
  119. static unsigned int         tNoOfOperations = 500;
  120. static unsigned int         tSleepTime = 0;
  121. static unsigned int         tNoOfLongPK = 1;
  122. static unsigned int         tSizeOfLongPK = 1;
  123. static unsigned int         t_instances = 1;
  125. //Program Flags
  126. static int                  theSimpleFlag = 0;
  127. static int                  theDirtyFlag = 0;
  128. static int                  theWriteFlag = 0;
  129. static int                  theStdTableNameFlag = 0;
  130. static int                  theTableCreateFlag = 0;
  131. static bool                 theTempTable = false;
  132. static bool                 VerifyFlag = true;
  133. static bool                 useLongKeys = false;
  134. static bool                 verbose = false;
  135. #ifdef USE_MYSQL
  136. static bool                 use_ndb = false;
  137. static int                  engine_id = 0;
  138. static int                  sockets[16];
  139. static int                  n_sockets = 0;
  140. static char*                engine[] =
  141.   { 
  142.     " ENGINE = NDBCLUSTER ", // use default engine
  143.     " ENGINE = MEMORY ",
  144.     " ENGINE = MYISAM ",
  145.     " ENGINE = INNODB "
  146.   };
  147. #else
  148. static bool                 use_ndb = true;
  149. #endif
  151. static ErrorData theErrorData; // Part of flexBench-program
  153. #define START_TIMER { NdbTimer timer; timer.doStart();
  154. #define STOP_TIMER timer.doStop();
  155. #define PRINT_TIMER(text, trans, opertrans) timer.printTransactionStatistics(text, trans, opertrans); };
  157. #include <NdbTCP.h>
  159. #ifdef CEBIT_STAT
  160. #include <NdbMutex.h>
  161. static bool statEnable = false;
  162. static char statHost[100];
  163. static int statFreq = 100;
  164. static int statPort = 0;
  165. static int statSock = -1;
  166. static enum { statError = -1, statClosed, statOpen } statState;
  167. static NdbMutex statMutex = NDB_MUTEX_INITIALIZER;
  168. #endif
  170. //-------------------------------------------------------------------
  171. // Statistical Reporting routines
  172. //-------------------------------------------------------------------
  173. #ifdef CEBIT_STAT
  174. // Experimental client-side statistic for CeBIT
  176. static void
  177. statReport(enum StartType st, int ops)
  178. {
  179.   if (!statEnable)
  180.     return;
  181.   if (NdbMutex_Lock(&statMutex) < 0) {
  182.     if (statState != statError) {
  183.       ndbout_c("stat: lock mutex failed: %s", strerror(errno));
  184.       statState = statError;
  185.     }
  186.     return;
  187.   }
  188.   static int nodeid;
  189.   // open connection
  190.   if (statState != statOpen) {
  191.     char *p = getenv("NDB_NODEID"); // ndbnet sets NDB_NODEID
  192.     nodeid = p == 0 ? 0 : atoi(p);
  193.     if ((statSock = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
  194.       if (statState != statError) {
  195. ndbout_c("stat: create socket failed: %s", strerror(errno));
  196. statState = statError;
  197.       }
  198.       (void)NdbMutex_Unlock(&statMutex);
  199.       return;
  200.     }
  201.     struct sockaddr_in saddr;
  202.     memset(&saddr, 0, sizeof(saddr));
  203.     saddr.sin_family = AF_INET;
  204.     saddr.sin_port = htons(statPort);
  205.     if (Ndb_getInAddr(&saddr.sin_addr, statHost) < 0) {
  206.       if (statState != statError) {
  207. ndbout_c("stat: host %s not found", statHost);
  208. statState = statError;
  209.       }
  210.       (void)close(statSock);
  211.       (void)NdbMutex_Unlock(&statMutex);
  212.       return;
  213.     }
  214.     if (connect(statSock, (struct sockaddr *)&saddr, sizeof(saddr)) < 0) {
  215.       if (statState != statError) {
  216. ndbout_c("stat: connect failed: %s", strerror(errno));
  217. statState = statError;
  218.       }
  219.       (void)close(statSock);
  220.       (void)NdbMutex_Unlock(&statMutex);
  221.       return;
  222.     }
  223.     statState = statOpen;
  224.     ndbout_c("stat: connection to %s:%d opened", statHost, (int)statPort);
  225.   }
  226.   const char *text;
  227.   switch (st) {
  228.   case stInsert:
  229.     text = "insert";
  230.     break;
  231.   case stVerify:
  232.     text = "verify";
  233.     break;
  234.   case stRead:
  235.     text = "read";
  236.     break;
  237.   case stUpdate:
  238.     text = "update";
  239.     break;
  240.   case stDelete:
  241.     text = "delete";
  242.     break;
  243.   case stVerifyDelete:
  244.     text = "verifydelete";
  245.     break;
  246.   default:
  247.     text = "unknown";
  248.     break;
  249.   }
  250.   char buf[100];
  251.   sprintf(buf, "%d %s %dn", nodeid, text, ops);
  252.   int len = strlen(buf);
  253.   // assume SIGPIPE already ignored
  254.   if (write(statSock, buf, len) != len) {
  255.     if (statState != statError) {
  256.       ndbout_c("stat: write failed: %s", strerror(errno));
  257.       statState = statError;
  258.     }
  259.     (void)close(statSock);
  260.     (void)NdbMutex_Unlock(&statMutex);
  261.     return;
  262.   }
  263.   (void)NdbMutex_Unlock(&statMutex);
  264. }
  265. #endif // CEBIT_STAT
  267. static void 
  268. resetThreads(ThreadData* pt){
  269.   for (unsigned int i = 0; i < tNoOfThreads; i++){
  270.     pt[i].threadReady = 0;
  271.     pt[i].threadResult = 0;
  272.     pt[i].threadStart = stIdle;
  273.   }
  274. }
  276. static int 
  277. checkThreadResults(ThreadData* pt){
  278.   for (unsigned int i = 0; i < tNoOfThreads; i++){
  279.     if(pt[i].threadResult != 0){
  280.       ndbout_c("Thread%d reported fatal error %d", i, pt[i].threadResult);
  281.       return -1;
  282.     }
  283.   }
  284.   return 0;
  285. }
  287. static
  288. void 
  289. waitForThreads(ThreadData* pt)
  290. {
  291.   int cont = 1;
  292.   while (cont){
  293.     NdbSleep_MilliSleep(100);
  294.     cont = 0;
  295.     for (unsigned int i = 0; i < tNoOfThreads; i++){
  296.       if (pt[i].threadReady == 0) 
  297.     cont = 1;
  298.     }
  299.   }
  300. }
  302. static void 
  303. tellThreads(ThreadData* pt, StartType what)
  304. {
  305.   for (unsigned int i = 0; i < tNoOfThreads; i++) 
  306.     pt[i].threadStart = what;
  307. }
  309. NDB_COMMAND(flexBench, "flexBench", "flexBench", "flexbench", 65535)
  310. {
  311.   ndb_init();
  312.   ThreadData*           pThreadsData;
  313.   int                   tLoops = 0;
  314.   int                   returnValue = NDBT_OK;
  315.   if (readArguments(argc, argv) != 0){
  316.     input_error();
  317.     return NDBT_ProgramExit(NDBT_WRONGARGS);
  318.   }
  319.   NdbAutoPtr<char> p10;
  320.   if(useLongKeys){
  321.     int e1 = sizeof(char*) * tNoOfLongPK;
  322.     int e2_1 = strlen("KEYATTR  ") + 1;
  323.     int e2 = e2_1 * tNoOfLongPK;
  324.     char *tmp = (char *) malloc(e1 + e2);
  325.     p10.reset(tmp);
  326.     longKeyAttrName = (char **) tmp;
  327.     tmp += e1;
  328.     for (Uint32 i = 0; i < tNoOfLongPK; i++) {
  329.       //      longKeyAttrName[i] = (char *) malloc(strlen("KEYATTR  ") + 1);
  330.       longKeyAttrName[i] = tmp;
  331.       tmp += e2_1;
  332.       memset(longKeyAttrName[i], 0, e2_1);
  333.       sprintf(longKeyAttrName[i], "KEYATTR%i", i);
  334.     }
  335.   }
  337.   NdbAutoObjArrayPtr<ThreadData>
  338.     p12( pThreadsData = new ThreadData[tNoOfThreads] );
  340.  
  341.   ndbout << endl << "FLEXBENCH - Starting normal mode" << endl;
  342.   ndbout << "Perform benchmark of insert, update and delete transactions"<< endl;
  343.   ndbout << "  " << tNoOfThreads << " thread(s) " << endl;
  344.   ndbout << "  " << tNoOfLoops << " iterations " << endl;
  345.   ndbout << "  " << tNoOfTables << " table(s) and " << 1 << " operation(s) per transaction " <<endl;
  346.   ndbout << "  " << tNoOfAttributes << " attributes per table " << endl;
  347.   ndbout << "  " << tNoOfOperations << " transaction(s) per thread and round " << endl;
  348.   ndbout << "  " << tAttributeSize << " is the number of 32 bit words per attribute "<< endl;
  349.   ndbout << "  " << "Table(s) without logging: " << (Uint32)theTempTable << endl;
  350.   
  351.   if(useLongKeys)
  352.     ndbout << "  " << "Using long keys with " << tNoOfLongPK << " keys a' " << 
  353.       tSizeOfLongPK * 4 << " bytes each." << endl;
  354.   
  355.   ndbout << "  " << "Verification is " ; 
  356.   if(VerifyFlag) {
  357.       ndbout << "enabled" << endl ;
  358.   }else{
  359.       ndbout << "disabled" << endl ;
  360.   }
  361.   if (use_ndb) {
  362.     ndbout << "Use NDB API with NdbPool in this test case" << endl;
  363.     ndbout << "Pool size = " << t_instances << endl;
  364.   } else {
  365.     ndbout << "Use mysql client with " << engine[engine_id];
  366.     ndbout << " as engine" << endl;
  367.   }
  368.   theErrorData.printSettings(ndbout);
  369.   
  370.   NdbThread_SetConcurrencyLevel(tNoOfThreads + 2);
  372. #ifdef USE_MYSQL
  373.   MYSQL mysql;
  374.   if (!use_ndb) {
  375.     if ( mysql_thread_safe() == 0 ) {
  376.       ndbout << "Not thread safe mysql library..." << endl;
  377.       return NDBT_ProgramExit(NDBT_FAILED);
  378.     }
  380.     ndbout << "Connecting to MySQL..." <<endl;
  382.     mysql_init(&mysql);
  383.     {
  384.       int the_socket = sockets[0];
  385.       char the_socket_name[1024];
  386.       sprintf(the_socket_name, "%s%u%s", "/tmp/mysql.",the_socket,".sock");
  387.       //    sprintf(the_socket_name, "%s", "/tmp/mysql.sock");
  388.       ndbout << the_socket_name << endl;
  389.       if ( mysql_real_connect(&mysql,
  390.     "localhost",
  391.     "root",
  392.     "",
  393.     "test",
  394.     the_socket,
  395.     the_socket_name,
  396.     0) == NULL ) {
  397.         ndbout << "Connect failed" <<endl;
  398.         returnValue = NDBT_FAILED;
  399.       }
  400.     }
  401.     if(returnValue == NDBT_OK){
  402.       mysql_set_server_option(&mysql, MYSQL_OPTION_MULTI_STATEMENTS_ON);
  403.       if (createTables(&mysql) != 0){
  404.         returnValue = NDBT_FAILED;
  405.       }
  406.     }
  407.   }
  408. #endif
  409.   if (use_ndb) {
  410.     Uint32 ndb_id = 0;
  411.     if (!create_instance(t_instances, 1, t_instances)) {
  412.       ndbout << "Creation of the NdbPool failed" << endl;
  413.       returnValue = NDBT_FAILED;
  414.     } else {
  415.       Ndb* pNdb = get_ndb_object(ndb_id, "test", "def");
  416.       if (pNdb == NULL) {
  417.         ndbout << "Failed to get a NDB object" << endl;
  418.         returnValue = NDBT_FAILED;
  419.       } else {
  420.         tNodeId = pNdb->getNodeId();
  421.         ndbout << "  NdbAPI node with id = " << tNodeId << endl;
  422.         ndbout << endl;
  423.   
  424.         ndbout << "Waiting for ndb to become ready..." <<endl;
  425.         if (pNdb->waitUntilReady(2000) != 0){
  426.           ndbout << "NDB is not ready" << endl;
  427.           ndbout << "Benchmark failed!" << endl;
  428.           returnValue = NDBT_FAILED;
  429.         }
  430.         if(returnValue == NDBT_OK){
  431.           if (createTables(pNdb) != 0){
  432.             returnValue = NDBT_FAILED;
  433.           }
  434.         }
  435.         return_ndb_object(pNdb, ndb_id);
  436.       }
  437.     }
  438.   }
  439.   if(returnValue == NDBT_OK){
  441.     sleepBeforeStartingTest(tSleepTime);
  442.     
  443.     /****************************************************************
  444.      *  Create threads.                                           *
  445.      ****************************************************************/
  446.     resetThreads(pThreadsData);
  447.     
  448.     for (unsigned int i = 0; i < tNoOfThreads; i++){  
  449.       pThreadsData[i].threadNo = i;
  450.       pThreadsData[i].threadLife = NdbThread_Create(flexBenchThread,
  451.                                                     (void**)&pThreadsData[i],
  452.                                                     32768,
  453.                                                     "flexBenchThread",
  454.                                                     NDB_THREAD_PRIO_LOW);
  455.     }
  456.     
  457.     waitForThreads(pThreadsData);
  458.     
  459.     ndbout << endl <<  "All threads started" << endl << endl;
  460.     
  461.     /****************************************************************
  462.      * Execute program.                                             *
  463.      ****************************************************************/
  464.   
  465.     for(;;){
  467.       int loopCount = tLoops + 1;
  468.       ndbout << endl << "Loop # " << loopCount  << endl << endl;
  469.       
  470.       /****************************************************************
  471.        * Perform inserts.                                             *
  472.        ****************************************************************/
  473.       // Reset and start timer
  474.       START_TIMER;
  475.       // Give insert-command to all threads
  476.       resetThreads(pThreadsData);
  477.       tellThreads(pThreadsData, stInsert);
  478.       waitForThreads(pThreadsData);
  479.       if (checkThreadResults(pThreadsData) != 0){
  480.         ndbout << "Error: Threads failed in performing insert" << endl;
  481.         returnValue = NDBT_FAILED;
  482.         break;
  483.       }        
  484.       // stop timer and print results.
  485.       STOP_TIMER;
  486.       PRINT_TIMER("insert", tNoOfOperations*tNoOfThreads, tNoOfTables);
  487.       /****************************************************************
  488.       * Verify inserts.                                             *
  489.       ****************************************************************/
  490.       if (VerifyFlag) {
  491. resetThreads(pThreadsData);
  492. ndbout << "Verifying inserts...t" ;
  493. tellThreads(pThreadsData, stVerify);
  494. waitForThreads(pThreadsData);
  495. if (checkThreadResults(pThreadsData) != 0){
  496.   ndbout << "Error: Threads failed while verifying inserts" << endl;
  497.   returnValue = NDBT_FAILED;
  498.   break;
  499. }else{
  500.   ndbout << "ttOK" << endl << endl ;
  501. }
  502.       }
  503.       
  504.       /****************************************************************
  505.        * Perform read.                                                *
  506.        ****************************************************************/
  507.       // Reset and start timer 
  508.       START_TIMER;
  509.       // Give read-command to all threads
  510.       resetThreads(pThreadsData);
  511.       tellThreads(pThreadsData, stRead);
  512.       waitForThreads(pThreadsData);
  513.       if (checkThreadResults(pThreadsData) != 0){
  514.         ndbout << "Error: Threads failed in performing read" << endl;
  515.         returnValue = NDBT_FAILED;
  516.         break;
  517.       }
  518.       // stop timer and print results.
  519.       STOP_TIMER;
  520.       PRINT_TIMER("read", tNoOfOperations*tNoOfThreads, tNoOfTables);
  521.       
  522.       /****************************************************************
  523.        * Perform update.                                              *
  524.        ****************************************************************/
  525.       // Reset and start timer
  526.       START_TIMER;
  527.       // Give update-command to all threads
  528.       resetThreads(pThreadsData);
  529.       tellThreads(pThreadsData, stUpdate);
  530.       waitForThreads(pThreadsData);
  531.       if (checkThreadResults(pThreadsData) != 0){
  532.         ndbout << "Error: Threads failed in performing update" << endl;
  533.         returnValue = NDBT_FAILED;
  534.         break;
  535.       }
  536.       // stop timer and print results.
  537.       STOP_TIMER;
  538.       PRINT_TIMER("update", tNoOfOperations*tNoOfThreads, tNoOfTables);
  539.       
  540.       /****************************************************************
  541.       * Verify updates.                                             *
  542.       ****************************************************************/
  543.       if (VerifyFlag) {
  544. resetThreads(pThreadsData);
  545. ndbout << "Verifying updates...t" ;
  546. tellThreads(pThreadsData, stVerify);
  547. waitForThreads(pThreadsData);
  548. if (checkThreadResults(pThreadsData) != 0){
  549.   ndbout << "Error: Threads failed while verifying updates" << endl;
  550.   returnValue = NDBT_FAILED;
  551.   break;
  552. }else{
  553.           ndbout << "ttOK" << endl << endl ;
  554. }
  555.       }
  556.       
  557.       /****************************************************************
  558.        * Perform read.                                             *
  559.        ****************************************************************/
  560.       // Reset and start timer
  561.       START_TIMER;
  562.       // Give read-command to all threads
  563.       resetThreads(pThreadsData);
  564.       tellThreads(pThreadsData, stRead);
  565.       waitForThreads(pThreadsData);
  566.       if (checkThreadResults(pThreadsData) != 0){
  567.         ndbout << "Error: Threads failed in performing read" << endl;
  568.         returnValue = NDBT_FAILED;
  569.         break;
  570.       }
  571.       // stop timer and print results.
  572.       STOP_TIMER;
  573.       PRINT_TIMER("read", tNoOfOperations*tNoOfThreads, tNoOfTables);
  575.       /****************************************************************
  576.        * Perform delete.                                              *
  577.        ****************************************************************/
  578.       // Reset and start timer
  579.       START_TIMER;
  580.       // Give delete-command to all threads
  581.       resetThreads(pThreadsData);
  582.       tellThreads(pThreadsData, stDelete);
  583.       waitForThreads(pThreadsData);
  584.       if (checkThreadResults(pThreadsData) != 0){
  585.         ndbout << "Error: Threads failed in performing delete" << endl;
  586.         returnValue = NDBT_FAILED;
  587.         break;
  588.       }
  589.       // stop timer and print results.
  590.       STOP_TIMER;
  591.       PRINT_TIMER("delete", tNoOfOperations*tNoOfThreads, tNoOfTables);
  593.       /****************************************************************
  594.       * Verify deletes.                                              *
  595.       ****************************************************************/
  596.       if (VerifyFlag) {
  597. resetThreads(pThreadsData);
  598. ndbout << "Verifying tuple deletion..." ;
  599. tellThreads(pThreadsData, stVerifyDelete);
  600. waitForThreads(pThreadsData);
  601. if (checkThreadResults(pThreadsData) != 0){
  602.           ndbout << "Error: Threads failed in verifying deletes" << endl;
  603.           returnValue = NDBT_FAILED;
  604.           break;
  605. }else{ 
  606.           ndbout << "ttOK" << endl << endl ;
  607. }
  608.       }
  609.       
  610.       ndbout << "--------------------------------------------------" << endl;
  612.       tLoops++;
  614.       if ( 0 != tNoOfLoops && tNoOfLoops <= tLoops ) 
  615.         break;
  616.       theErrorData.printErrorCounters();
  617.     }
  618.     
  619.     resetThreads(pThreadsData);
  620.     tellThreads(pThreadsData, stStop);
  621.     waitForThreads(pThreadsData);
  623.     void * tmp;
  624.     for(Uint32 i = 0; i<tNoOfThreads; i++){
  625.       NdbThread_WaitFor(pThreadsData[i].threadLife, &tmp);
  626.       NdbThread_Destroy(&pThreadsData[i].threadLife);
  627.     }
  628.   }
  629. #ifdef USE_MYSQL
  630.   if (!use_ndb) {
  631.     dropTables(&mysql);
  632.     mysql_close(&mysql);
  633.   }
  634. #endif
  635.   if (use_ndb) {
  636.     drop_instance();
  637.   }
  638.   theErrorData.printErrorCounters();
  639.   return NDBT_ProgramExit(returnValue);
  640. }
  641. ////////////////////////////////////////
  644. unsigned long get_hash(unsigned long * hash_key, int len)
  645. {
  646.   unsigned long hash_value = 147;
  647.   unsigned h_key;
  648.   int i;
  649.   for (i = 0; i < len; i++)
  650.     {
  651.       h_key = hash_key[i];
  652.       hash_value = (hash_value << 5) + hash_value + (h_key & 255);
  653.       hash_value = (hash_value << 5) + hash_value + ((h_key >> 8) & 255);
  654.       hash_value = (hash_value << 5) + hash_value + ((h_key >> 16) & 255);
  655.       hash_value = (hash_value << 5) + hash_value + ((h_key >> 24) & 255);
  656.     }
  657.   return hash_value;
  658. }
  660. // End of warming up phase
  664. static void* flexBenchThread(void* pArg)
  665. {
  666.   ThreadData*       pThreadData = (ThreadData*)pArg;
  667.   unsigned int      threadNo, threadBase;
  668.   Ndb*              pNdb = NULL ;
  669.   Uint32            ndb_id = 0;
  670.   NdbConnection     *pTrans = NULL ;
  671.   NdbOperation**    pOps = NULL ;
  672.   StartType         tType ;
  673.   StartType         tSaveType ;
  674.   NdbRecAttr*       tTmp = NULL ;
  675.   int*              attrValue = NULL ;
  676.   int*              attrRefValue = NULL ;
  677.   int               check = 0 ;
  678.   int               loopCountOps, loopCountTables, loopCountAttributes;
  679.   int               tAttemptNo = 0;
  680.   int               tRetryAttempts = 20;
  681.   int               tResult = 0;
  682.   int               tSpecialTrans = 0;
  683.   int               nRefLocalOpOffset = 0 ;
  684.   int               nReadBuffSize = 
  685.     tNoOfTables * tNoOfAttributes * sizeof(int) * tAttributeSize ;
  686.   int               nRefBuffSize = 
  687.     tNoOfOperations * tNoOfAttributes * sizeof(int) * tAttributeSize ;
  688.   unsigned***           longKeyAttrValue = NULL;
  691.   threadNo = pThreadData->threadNo ;
  693. #ifdef USE_MYSQL
  694.   MYSQL mysql;
  695.   int the_socket = sockets[threadNo % n_sockets];
  696.   char the_socket_name[1024];
  697.   //sprintf(the_socket_name, "%s", "/tmp/mysql.sock");
  698.   sprintf(the_socket_name, "%s%u%s", "/tmp/mysql.",the_socket,".sock");
  699.   if (!use_ndb) {
  700.     ndbout << the_socket_name << endl;
  701.     ndbout << "Thread connecting to MySQL... " << endl;
  702.     mysql_init(&mysql);
  703.     
  704.     if ( mysql_real_connect(&mysql,
  705.     "localhost",
  706.     "root",
  707.     "",
  708.     "test",
  709.     the_socket,
  710.     the_socket_name,
  711.     0) == NULL ) {
  712.       ndbout << "failed" << endl;
  713.       return 0;
  714.     }
  715.     ndbout << "ok" << endl;
  717.     int r;
  718.     if (tNoOfTables > 1)
  719.       r = mysql_autocommit(&mysql, 0);
  720.     else
  721.       r = mysql_autocommit(&mysql, 1);
  723.     if (r) {
  724.       ndbout << "autocommit on/off failed" << endl;
  725.       return 0;
  726.     }
  727.   }
  728. #endif
  730.   NdbAutoPtr<int> p00( attrValue= (int*)malloc(nReadBuffSize) ) ;
  731.   NdbAutoPtr<int> p01( attrRefValue= (int*)malloc(nRefBuffSize) );
  732.   if (use_ndb) {
  733.     pOps = (NdbOperation**)malloc(tNoOfTables*sizeof(NdbOperation*)) ;
  734.   }
  735.   NdbAutoPtr<NdbOperation*> p02( pOps );
  737.   if( !attrValue || !attrRefValue ||
  738.       ( use_ndb && ( !pOps) ) ){
  739.     // Check allocations to make sure we got all the memory we asked for
  740.     ndbout << "One or more memory allocations failed when starting thread #";
  741.     ndbout << threadNo << endl ;
  742.     ndbout << "Thread #" << threadNo << " will now exit" << endl ;
  743.     tResult = 13 ;
  744.     return 0;
  745.   }
  746.   
  747.   if (use_ndb) {
  748.     pNdb = get_ndb_object(ndb_id, "test", "def");
  749.     if (pNdb == NULL) {
  750.       ndbout << "Failed to get an NDB object" << endl;
  751.       ndbout << "Thread #" << threadNo << " will now exit" << endl ;
  752.       tResult = 13;
  753.       return 0;
  754.     }
  755.     pNdb->waitUntilReady();
  756.     return_ndb_object(pNdb, ndb_id);
  757.     pNdb = NULL;
  758.   }
  760.   // To make sure that two different threads doesn't operate on the same record
  761.   // Calculate an "unique" number to use as primary key
  762.   threadBase = (threadNo * 2000000) + (tNodeId * 260000000);
  764.   NdbAutoPtr<char> p22;
  765.   if(useLongKeys){
  766.     // Allocate and populate the longkey array.
  767.     int e1 = sizeof(unsigned**) * tNoOfOperations;
  768.     int e2 = sizeof(unsigned*) * tNoOfLongPK * tNoOfOperations;
  769.     int e3 = sizeof(unsigned) * tSizeOfLongPK * tNoOfLongPK * tNoOfOperations;
  770.     char* tmp;
  771.     p22.reset(tmp = (char*)malloc(e1+e2+e3));
  773.     longKeyAttrValue = (unsigned ***) tmp;
  774.     tmp += e1;
  775.     for (Uint32 n = 0; n < tNoOfOperations; n++) {
  776.       longKeyAttrValue[n] = (unsigned **) tmp;
  777.       tmp += sizeof(unsigned*) * tNoOfLongPK;
  778.     }
  780.     for (Uint32 n = 0; n < tNoOfOperations; n++){
  781.       for (Uint32 i = 0; i < tNoOfLongPK ; i++) {
  782. longKeyAttrValue[n][i] = (unsigned *) tmp;
  783. tmp += sizeof(unsigned) * tSizeOfLongPK;
  784. memset(longKeyAttrValue[n][i], 0, sizeof(unsigned) * tSizeOfLongPK);
  785. for(Uint32 j = 0; j < tSizeOfLongPK; j++) {
  786.   // Repeat the unique value to fill up the long key.
  787.   longKeyAttrValue[n][i][j] = threadBase + n; 
  788. }
  789.       }
  790.     }
  791.   }
  793.   int nRefOpOffset = 0 ;
  794.   //Assign reference attribute values to memory
  795.   for(Uint32 ops = 1 ; ops < tNoOfOperations ; ops++){
  796.     // Calculate offset value before going into the next loop
  797.     nRefOpOffset = tAttributeSize*tNoOfAttributes*(ops-1) ; 
  798.     for(Uint32 a = 0 ; a < tNoOfAttributes ; a++){
  799.       *(int*)&attrRefValue[nRefOpOffset + tAttributeSize*a] = 
  800. (int)(threadBase + ops + a) ;
  801.     }
  802.   }
  804. #ifdef CEBIT_STAT
  805.   // ops not yet reported
  806.   int statOps = 0;
  807. #endif
  809. #ifdef USE_MYSQL
  810.   // temporary buffer to store prepared statement text
  811.   char buf[2048];
  812.   MYSQL_STMT** prep_read   = NULL;
  813.   MYSQL_STMT** prep_delete = NULL;
  814.   MYSQL_STMT** prep_update = NULL;
  815.   MYSQL_STMT** prep_insert = NULL;
  816.   MYSQL_BIND* bind_delete = NULL;
  817.   MYSQL_BIND* bind_read   = NULL;
  818.   MYSQL_BIND* bind_update = NULL;
  819.   MYSQL_BIND* bind_insert = NULL;
  820.   int* mysql_data = NULL;
  822.   NdbAutoPtr<char> p21;
  824.   if (!use_ndb) {
  825.     // data array to which prepared statements are bound
  826.     char* tmp;
  827.     int e1 = sizeof(int)*tAttributeSize*tNoOfAttributes;
  828.     int e2 = sizeof(MYSQL_BIND)*tNoOfAttributes;
  829.     int e3 = sizeof(MYSQL_BIND)*tNoOfAttributes;
  830.     int e4 = sizeof(MYSQL_BIND)*tNoOfAttributes;
  831.     int e5 = sizeof(MYSQL_BIND)*1;
  832.     int e6 = sizeof(MYSQL_STMT*)*tNoOfTables;
  833.     int e7 = sizeof(MYSQL_STMT*)*tNoOfTables;
  834.     int e8 = sizeof(MYSQL_STMT*)*tNoOfTables;
  835.     int e9 = sizeof(MYSQL_STMT*)*tNoOfTables;
  836.     p21.reset(tmp = (char*)malloc(e1+e2+e3+e4+e5+e6+e7+e8+e9));
  838.     mysql_data  = (int*)tmp;         tmp += e1;
  839.     bind_insert = (MYSQL_BIND*)tmp;  tmp += e2;
  840.     bind_update = (MYSQL_BIND*)tmp;  tmp += e3;
  841.     bind_read   = (MYSQL_BIND*)tmp;  tmp += e4;
  842.     bind_delete = (MYSQL_BIND*)tmp;  tmp += e5;
  843.     prep_insert = (MYSQL_STMT**)tmp; tmp += e6;
  844.     prep_update = (MYSQL_STMT**)tmp; tmp += e7;
  845.     prep_read   = (MYSQL_STMT**)tmp; tmp += e8;
  846.     prep_delete = (MYSQL_STMT**)tmp;
  848.     for (Uint32 ca = 0; ca < tNoOfAttributes; ca++){
  849.       MYSQL_BIND& bi = bind_insert[ca];
  850.       bi.buffer_type = MYSQL_TYPE_LONG;
  851.       bi.buffer = (char*)&mysql_data[ca*tAttributeSize];
  852.       bi.buffer_length = 0;
  853.       bi.length = NULL;
  854.       bi.is_null = NULL;
  855.     }//for
  856.     
  857.     for (Uint32 ca = 0; ca < tNoOfAttributes; ca++){
  858.       MYSQL_BIND& bi = bind_update[ca];
  859.       bi.buffer_type = MYSQL_TYPE_LONG;
  860.       if ( ca == tNoOfAttributes-1 ) // the primary key comes last in statement
  861. bi.buffer = (char*)&mysql_data[0];
  862.       else
  863. bi.buffer = (char*)&mysql_data[(ca+1)*tAttributeSize];
  864.       bi.buffer_length = 0;
  865.       bi.length = NULL;
  866.       bi.is_null = NULL;
  867.     }//for
  868.     
  869.     for (Uint32 ca = 0; ca < tNoOfAttributes; ca++){
  870.       MYSQL_BIND& bi = bind_read[ca];
  871.       bi.buffer_type = MYSQL_TYPE_LONG;
  872.       bi.buffer = (char*)&mysql_data[ca*tAttributeSize];
  873.       bi.buffer_length = 4;
  874.       bi.length = NULL;
  875.       bi.is_null = NULL;
  876.     }//for
  877.     
  878.     for (Uint32 ca = 0; ca < 1; ca++){
  879.       MYSQL_BIND& bi = bind_delete[ca];
  880.       bi.buffer_type = MYSQL_TYPE_LONG;
  881.       bi.buffer = (char*)&mysql_data[ca*tAttributeSize];
  882.       bi.buffer_length = 0;
  883.       bi.length = NULL;
  884.       bi.is_null = NULL;
  885.     }//for
  886.     
  887.     for (Uint32 i = 0; i < tNoOfTables; i++) {
  888.       int pos = 0;
  889.       pos += sprintf(buf+pos, "%s%s%s",
  890.      "INSERT INTO ",
  891.      tableName[i],
  892.      " VALUES(");
  893.       pos += sprintf(buf+pos, "%s", "?");
  894.       for (Uint32 j = 1; j < tNoOfAttributes; j++) {
  895. pos += sprintf(buf+pos, "%s", ",?");
  896.       }
  897.       pos += sprintf(buf+pos, "%s", ")");
  898.       if (verbose)
  899. ndbout << buf << endl;
  900.       prep_insert[i] = mysql_prepare(&mysql, buf, pos);
  901.       if (prep_insert[i] == 0) {
  902. ndbout << "mysql_prepare: " << mysql_error(&mysql) << endl;
  903. return 0;
  904.       }
  905.       if (mysql_bind_param(prep_insert[i], bind_insert)) {
  906. ndbout << "mysql_bind_param: " << mysql_error(&mysql) << endl;
  907. return 0;
  908.       }
  909.     }
  910.     
  911.     for (Uint32 i = 0; i < tNoOfTables; i++) {
  912.       int pos = 0;
  913.       pos += sprintf(buf+pos, "%s%s%s",
  914.      "UPDATE ",
  915.      tableName[i],
  916.      " SET ");
  917.       for (Uint32 j = 1; j < tNoOfAttributes; j++) {
  918. if (j != 1)
  919.   pos += sprintf(buf+pos, "%s", ",");
  920. pos += sprintf(buf+pos, "%s%s", attrName[j],"=?");
  921.       }
  922.       pos += sprintf(buf+pos, "%s%s%s", " WHERE ", attrName[0], "=?");
  923.       
  924.       if (verbose)
  925. ndbout << buf << endl;
  926.       prep_update[i] = mysql_prepare(&mysql, buf, pos);
  927.       if (prep_update[i] == 0) {
  928. ndbout << "mysql_prepare: " << mysql_error(&mysql) << endl;
  929. return 0;
  930.       }
  931.       if (mysql_bind_param(prep_update[i], bind_update)) {
  932. ndbout << "mysql_bind_param: " << mysql_error(&mysql) << endl;
  933. return 0;
  934.       }
  935.     }
  936.     
  937.     for (Uint32 i = 0; i < tNoOfTables; i++) {
  938.       int pos = 0;
  939.       pos += sprintf(buf+pos, "%s", "SELECT ");
  940.       for (Uint32 j = 1; j < tNoOfAttributes; j++) {
  941. if (j != 1)
  942.   pos += sprintf(buf+pos, "%s", ",");
  943. pos += sprintf(buf+pos, "%s", attrName[j]);
  944.       }
  945.       pos += sprintf(buf+pos, "%s%s%s%s%s",
  946.      " FROM ",
  947.      tableName[i],
  948.      " WHERE ",
  949.      attrName[0],
  950.      "=?");
  951.       if (verbose)
  952. ndbout << buf << endl;
  953.       prep_read[i] = mysql_prepare(&mysql, buf, pos);
  954.       if (prep_read[i] == 0) {
  955. ndbout << "mysql_prepare: " << mysql_error(&mysql) << endl;
  956. return 0;
  957.       }
  958.       if (mysql_bind_param(prep_read[i], bind_read)) {
  959. ndbout << "mysql_bind_param: " << mysql_error(&mysql) << endl;
  960. return 0;
  961.       }
  962.       if (mysql_bind_result(prep_read[i], &bind_read[1])) {
  963. ndbout << "mysql_bind_result: " << mysql_error(&mysql) << endl;
  964. return 0;
  965.       }
  966.     }
  967.     
  968.     for (Uint32 i = 0; i < tNoOfTables; i++) {
  969.       int pos = 0;
  970.       pos += sprintf(buf+pos, "%s%s%s%s%s",
  971.      "DELETE FROM ",
  972.      tableName[i],
  973.      " WHERE ",
  974.      attrName[0],
  975.      "=?");
  976.       if (verbose)
  977. ndbout << buf << endl;
  978.       prep_delete[i] = mysql_prepare(&mysql, buf, pos);
  979.       if (prep_delete[i] == 0) {
  980. ndbout << "mysql_prepare: " << mysql_error(&mysql) << endl;
  981. return 0;
  982.       }
  983.       if (mysql_bind_param(prep_delete[i], bind_delete)) {
  984. ndbout << "mysql_bind_param: " << mysql_error(&mysql) << endl;
  985. return 0;
  986.       }
  987.     }
  988.   }
  989. #endif
  991.   for (;;) {
  992.     pThreadData->threadResult = tResult; // Report error to main thread, 
  993.     // normally tResult is set to 0
  994.     pThreadData->threadReady = 1;
  996.     while (pThreadData->threadStart == stIdle){
  997.       NdbSleep_MilliSleep(100);
  998.     }//while
  1000.     // Check if signal to exit is received
  1001.     if (pThreadData->threadStart == stStop){
  1002.       pThreadData->threadReady = 1;
  1003.       // ndbout_c("Thread%d is stopping", threadNo);
  1004.       // In order to stop this thread, the main thread has signaled
  1005.       // stStop, break out of the for loop so that destructors
  1006.       // and the proper exit functions are called
  1007.       break;
  1008.     }//if
  1010.     tType = pThreadData->threadStart;
  1011.     tSaveType = tType;
  1012.     pThreadData->threadStart = stIdle;
  1014.     // Start transaction, type of transaction
  1015.     // is received in the array ThreadStart
  1016.     loopCountOps = tNoOfOperations;
  1017.     loopCountTables = tNoOfTables;
  1018.     loopCountAttributes = tNoOfAttributes;
  1019.     for (int count = 1; count < loopCountOps && tResult == 0;){
  1021.       if (use_ndb) {
  1022.         pNdb = get_ndb_object(ndb_id, "test", "def");
  1023.         if (pNdb == NULL) {
  1024.           ndbout << "Could not get Ndb object in thread" << threadNo;
  1025.           ndbout << endl;
  1026.           tResult = 1; //Indicate fatal error
  1027.           break;
  1028.         }
  1029. pTrans = pNdb->startTransaction();
  1030. if (pTrans == NULL) {
  1031.   // This is a fatal error, abort program
  1032.   ndbout << "Could not start transaction in thread" << threadNo;
  1033.   ndbout << endl;
  1034.   ndbout << pNdb->getNdbError() << endl;
  1035.   tResult = 1; // Indicate fatal error
  1036.   break; // Break out of for loop
  1037. }
  1038.       }
  1040.       // Calculate the current operation offset in the reference array
  1041.       nRefLocalOpOffset = tAttributeSize*tNoOfAttributes*(count - 1) ;
  1042.       int* tmpAttrRefValue = attrRefValue + nRefLocalOpOffset;
  1044.       for (int countTables = 0;
  1045.    countTables < loopCountTables && tResult == 0;
  1046.    countTables++) {
  1048. int nTableOffset = tAttributeSize *
  1049.   loopCountAttributes *
  1050.   countTables ;
  1052. int* tmpAttrValue = attrValue + nTableOffset;
  1054. if (use_ndb) {
  1055.   pOps[countTables] = pTrans->getNdbOperation(tableName[countTables]); 
  1056.   if (pOps[countTables] == NULL) {
  1057.     // This is a fatal error, abort program
  1058.     ndbout << "getNdbOperation: " << pTrans->getNdbError();
  1059.     tResult = 2; // Indicate fatal error
  1060.     break;
  1061.   }//if
  1063.   switch (tType) {
  1064.   case stInsert:          // Insert case
  1065.     if (theWriteFlag == 1 && theDirtyFlag == 1)
  1066.     pOps[countTables]->dirtyWrite();
  1067.     else if (theWriteFlag == 1)
  1068.       pOps[countTables]->writeTuple();
  1069.     else
  1070.       pOps[countTables]->insertTuple();
  1071.     break;
  1072.   case stRead:            // Read Case
  1073.     if (theSimpleFlag == 1)
  1074.       pOps[countTables]->simpleRead();
  1075.     else if (theDirtyFlag == 1)
  1076.       pOps[countTables]->dirtyRead();
  1077.     else
  1078.       pOps[countTables]->readTuple();
  1079.     break;
  1080.   case stUpdate:          // Update Case
  1081.     if (theWriteFlag == 1 && theDirtyFlag == 1)
  1082.       pOps[countTables]->dirtyWrite();
  1083.     else if (theWriteFlag == 1)
  1084.       pOps[countTables]->writeTuple();
  1085.     else if (theDirtyFlag == 1)
  1086.       pOps[countTables]->dirtyUpdate();
  1087.     else
  1088.       pOps[countTables]->updateTuple();
  1089.     break;
  1090.   case stDelete:          // Delete Case
  1091.     pOps[countTables]->deleteTuple();
  1092.     break;
  1093.   case stVerify:
  1094.     pOps[countTables]->readTuple();
  1095.     break;
  1096.   case stVerifyDelete:
  1097.     pOps[countTables]->readTuple();
  1098.     break;
  1099.   default:
  1100.     assert(false);
  1101.   }//switch
  1102.   
  1103.   if(useLongKeys){
  1104.     // Loop the equal call so the complete key is send to the kernel.
  1105.     for(Uint32 i = 0; i < tNoOfLongPK; i++) 
  1106.       pOps[countTables]->equal(longKeyAttrName[i], 
  1107.        (char *)longKeyAttrValue[count - 1][i],
  1108.        tSizeOfLongPK*4); 
  1109.   }
  1110.   else 
  1111.     pOps[countTables]->equal((char*)attrName[0], 
  1112.      (char*)&tmpAttrRefValue[0]);
  1114.   if (tType == stInsert) {
  1115.     for (int ca = 1; ca < loopCountAttributes; ca++){
  1116.       pOps[countTables]->setValue((char*)attrName[ca],
  1117.   (char*)&tmpAttrRefValue[tAttributeSize*ca]);
  1118.     }//for
  1119.   } else if (tType == stUpdate) {
  1120.     for (int ca = 1; ca < loopCountAttributes; ca++){
  1121.       int* tmp = (int*)&tmpAttrRefValue[tAttributeSize*ca];
  1122.       if (countTables == 0)
  1123. (*tmp)++;
  1124.       pOps[countTables]->setValue((char*)attrName[ca],(char*)tmp);
  1125.     }//for
  1126.   } else if (tType == stRead || stVerify == tType) {
  1127.     for (int ca = 1; ca < loopCountAttributes; ca++) {
  1128.       tTmp =
  1129. pOps[countTables]->getValue((char*)attrName[ca], 
  1130.     (char*)&tmpAttrValue[tAttributeSize*ca]);
  1131.     }//for
  1132.   } else if (stVerifyDelete == tType) {
  1133.     if(useLongKeys){
  1134.       tTmp = pOps[countTables]->getValue(longKeyAttrName[0], 
  1135.  (char*)&tmpAttrValue[0]);
  1136.     } else {
  1137.       tTmp = pOps[countTables]->getValue((char*)attrName[0], 
  1138.  (char*)&tmpAttrValue[0]);
  1139.     }
  1140.   }//if
  1141. } else { // !use_ndb
  1142. #ifndef USE_MYSQL
  1143.   assert(false);
  1144. #else
  1145.   switch (tType)
  1146.     {
  1147.     case stInsert:
  1148.       for (int ca = 0; ca < loopCountAttributes; ca++){
  1149. mysql_data[ca] = tmpAttrRefValue[tAttributeSize*ca];
  1150.       }//for
  1151.       if (mysql_execute(prep_insert[countTables])) {
  1152. ndbout << tableName[countTables];  
  1153. ndbout << " mysql_execute: " << mysql_error(&mysql) << endl;
  1154. tResult = 1 ;
  1155.       }
  1156.       break;
  1157.     case stUpdate:          // Update Case
  1158.       mysql_data[0] = tmpAttrRefValue[0];
  1159.       for (int ca = 1; ca < loopCountAttributes; ca++){
  1160. int* tmp = (int*)&tmpAttrRefValue[tAttributeSize*ca];
  1161. if (countTables == 0)
  1162.   (*tmp)++;
  1163. mysql_data[ca] = *tmp;
  1164.       }//for
  1165.       if (mysql_execute(prep_update[countTables])) {
  1166. ndbout << tableName[countTables];  
  1167. ndbout << " mysql_execute: " << mysql_error(&mysql) << endl;
  1168. tResult = 2 ;
  1169.       }
  1170.       break;
  1171.     case stVerify:
  1172.     case stRead:            // Read Case
  1173.       mysql_data[0] = tmpAttrRefValue[0];
  1174.       if (mysql_execute(prep_read[countTables])) {
  1175. ndbout << tableName[countTables];  
  1176. ndbout << " mysql_execute: " << mysql_error(&mysql) << endl;
  1177. tResult = 3 ;
  1178. break;
  1179.       }
  1180.       if (mysql_stmt_store_result(prep_read[countTables])) {
  1181. ndbout << tableName[countTables];  
  1182. ndbout << " mysql_stmt_store_result: "
  1183.        << mysql_error(&mysql) << endl;
  1184. tResult = 4 ;
  1185. break;
  1186.       }
  1187.       {
  1188. int rows= 0;
  1189. int r;
  1190. while ( (r= mysql_fetch(prep_read[countTables])) == 0 ){
  1191.   rows++;
  1192. }
  1193. if ( r == 1 ) {
  1194.   ndbout << tableName[countTables];  
  1195.   ndbout << " mysql_fetch: " << mysql_error(&mysql) << endl;
  1196.   tResult = 5 ;
  1197.   break;
  1198. }
  1199. if ( rows != 1 ) {
  1200.   ndbout << tableName[countTables];  
  1201.   ndbout << " mysql_fetch: rows = " << rows << endl;
  1202.   tResult = 6 ;
  1203.   break;
  1204. }
  1205.       }
  1206.       {
  1207. for (int ca = 1; ca < loopCountAttributes; ca++) {
  1208.   tmpAttrValue[tAttributeSize*ca] = mysql_data[ca];
  1209. }
  1210.       }
  1211.       break;
  1212.     case stDelete:          // Delete Case
  1213.       mysql_data[0] = tmpAttrRefValue[0];
  1214.       if (mysql_execute(prep_delete[countTables])) {
  1215. ndbout << tableName[countTables];  
  1216. ndbout << " mysql_execute: " << mysql_error(&mysql) << endl;
  1217. tResult = 7 ;
  1218. break;
  1219.       }
  1220.       break;
  1221.     case stVerifyDelete:
  1222.       {
  1223. sprintf(buf, "%s%s%s",
  1224. "SELECT COUNT(*) FROM ",tableName[countTables],";");
  1225. if (mysql_query(&mysql, buf)) {
  1226.   ndbout << buf << endl;
  1227.   ndbout << "Error: " << mysql_error(&mysql) << endl;
  1228.   tResult = 8 ;
  1229.   break;
  1230. }
  1231. MYSQL_RES *res = mysql_store_result(&mysql);
  1232. if ( res == NULL ) {
  1233.   ndbout << "mysql_store_result: "
  1234.  << mysql_error(&mysql) << endl
  1235.  << "errno: " << mysql_errno(&mysql) << endl;
  1236.   tResult = 9 ;
  1237.   break;
  1238. }
  1239. int num_fields = mysql_num_fields(res);
  1240. int num_rows   = mysql_num_rows(res);
  1241. if ( num_rows != 1 || num_fields != 1 ) {
  1242.   ndbout << tableName[countTables];  
  1243.   ndbout << " mysql_store_result: num_rows = " << num_rows
  1244.  << " num_fields = " << num_fields << endl;
  1245.   tResult = 10 ;
  1246.   break;
  1247. }
  1248. MYSQL_ROW row = mysql_fetch_row(res);
  1249. if ( row == NULL ) {
  1250.   ndbout << "mysql_fetch_row: "
  1251.  << mysql_error(&mysql) << endl;
  1252.   tResult = 11 ;
  1253.   break;
  1254. }
  1255. if ( *(char*)row[0] != '0' ) {
  1256.   ndbout << tableName[countTables];  
  1257.   ndbout << " mysql_fetch_row: value = "
  1258.  << (char*)(row[0]) << endl;
  1259.   tResult = 12 ;
  1260.   break;
  1261. }
  1262. mysql_free_result(res);
  1263.       }
  1264.       break;
  1265.     default:
  1266.       assert(false);
  1267.     }
  1268. #endif
  1269. }
  1270.       }//for Tables loop
  1272.       if (tResult != 0)
  1273. break;
  1275.       if (use_ndb){
  1276. check = pTrans->execute(Commit);
  1277.       } else {
  1278. #ifdef USE_MYSQL
  1279. if (tNoOfTables > 1)
  1280.   if (mysql_commit(&mysql)) {
  1281.     ndbout << " mysql_commit: " << mysql_error(&mysql) << endl;
  1282.     tResult = 13;
  1283.   } else 
  1284.     check = 0;
  1285. #endif
  1286.       }
  1288.       if (use_ndb) {
  1289. // Decide what kind of error this is
  1290. if ((tSpecialTrans == 1) &&
  1291.     (check == -1)) {
  1292. // --------------------------------------------------------------------
  1293. // A special transaction have been executed, change to check = 0 in
  1294. // certain situations.
  1295. // --------------------------------------------------------------------
  1296.   switch (tType) {
  1297.   case stInsert:          // Insert case
  1298.     if (630 == pTrans->getNdbError().code ) {
  1299.       check = 0;
  1300.       ndbout << "Insert with 4007 was successful" << endl;
  1301.     }//if
  1302.     break;
  1303.   case stDelete:          // Delete Case
  1304.     if (626 == pTrans->getNdbError().code ) {
  1305.       check = 0;
  1306.       ndbout << "Delete with 4007 was successful" << endl;
  1307.     }//if
  1308.     break;
  1309.   default:
  1310.     assert(false);
  1311.   }//switch
  1312. }//if
  1313. tSpecialTrans = 0;
  1314. if (check == -1) {
  1315.   if ((stVerifyDelete == tType) && 
  1316.       (626 == pTrans->getNdbError().code)) {
  1317.     // ----------------------------------------------
  1318.     // It's good news - the deleted tuple is gone, 
  1319.     // so reset "check" flag
  1320.     // ----------------------------------------------
  1321.     check = 0 ;
  1322.   } else {
  1323.     int retCode = 
  1324.       theErrorData.handleErrorCommon(pTrans->getNdbError());
  1325.     if (retCode == 1) {
  1326.       ndbout_c("execute: %d, %d, %s", count, tType, 
  1327.        pTrans->getNdbError().message );
  1328.       ndbout_c("Error code = %d", pTrans->getNdbError().code );
  1329.       tResult = 20;
  1330.     } else if (retCode == 2) {
  1331.       ndbout << "4115 should not happen in flexBench" << endl;
  1332.       tResult = 20;
  1333.     } else if (retCode == 3) {
  1334. // --------------------------------------------------------------------
  1335. // We are not certain if the transaction was successful or not.
  1336. // We must reexecute but might very well find that the transaction
  1337. // actually was updated. Updates and Reads are no problem here. Inserts
  1338. // will not cause a problem if error code 630 arrives. Deletes will
  1339. // not cause a problem if 626 arrives.
  1340. // --------------------------------------------------------------------
  1341.       if ((tType == stInsert) || (tType == stDelete)) {
  1342. tSpecialTrans = 1;
  1343.       }//if
  1344.     }//if
  1345.   }//if
  1346. }//if
  1347. // Check if retries should be made
  1348. if (check == -1 && tResult == 0) {
  1349.   if (tAttemptNo < tRetryAttempts){
  1350.     tAttemptNo++;
  1351.   } else {
  1352. // --------------------------------------------------------------------
  1353. // Too many retries have been made, report error and break out of loop
  1354. // --------------------------------------------------------------------
  1355.     ndbout << "Thread" << threadNo;
  1356.     ndbout << ": too many errors reported" << endl;
  1357.     tResult = 10;
  1358.     break;
  1359.   }//if            
  1360. }//if
  1361.       }
  1363.       if (check == 0){
  1364. // Go to the next record
  1365. count++;
  1366. tAttemptNo = 0;
  1367. #ifdef CEBIT_STAT
  1368. // report successful ops
  1369. if (statEnable) {
  1370.   statOps += loopCountTables;
  1371.   if (statOps >= statFreq) {
  1372.     statReport(tType, statOps);
  1373.     statOps = 0;
  1374.   }//if
  1375. }//if
  1376. #endif
  1377.       }//if
  1379.       if (stVerify == tType && 0 == check){
  1380. int nTableOffset = 0 ;
  1381. for (int a = 1 ; a < loopCountAttributes ; a++){
  1382.   for (int tables = 0 ; tables < loopCountTables ; tables++){
  1383.     nTableOffset = tables*loopCountAttributes*tAttributeSize;
  1384.     int ov =*(int*)&attrValue[nTableOffset + tAttributeSize*a];
  1385.     int nv =*(int*)&tmpAttrRefValue[tAttributeSize*a];
  1386.     if (ov != nv){
  1387.       ndbout << "Error in verify ";
  1388.       ndbout << "pk = " << tmpAttrRefValue[0] << ":" << endl;
  1389.       ndbout << "attrValue[" << nTableOffset + tAttributeSize*a << "] = " << ov << endl ;
  1390.       ndbout << "attrRefValue[" << nRefLocalOpOffset + tAttributeSize*a << "]" << nv << endl ;
  1391.       tResult = 11 ;
  1392.       break ;
  1393.     }//if
  1394.   }//for
  1395. }//for
  1396.       }// if(stVerify ... )
  1397.       if (use_ndb) {
  1398. pNdb->closeTransaction(pTrans);
  1399.         return_ndb_object(pNdb, ndb_id);
  1400.         pNdb = NULL;
  1401.       }
  1402.     }// operations loop
  1403. #ifdef CEBIT_STAT
  1404.     // report remaining successful ops
  1405.     if (statEnable) {
  1406.       if (statOps > 0) {
  1407. statReport(tType, statOps);
  1408. statOps = 0;
  1409.       }//if
  1410.     }//if
  1411. #endif
  1412.     if (pNdb) {
  1413.       pNdb->closeTransaction(pTrans);
  1414.       return_ndb_object(pNdb, ndb_id);
  1415.       pNdb = NULL;
  1416.     }
  1417.   }
  1419. #ifdef USE_MYSQL
  1420.   if (!use_ndb) {
  1421.     mysql_close(&mysql);
  1422.     for (Uint32 i = 0; i < tNoOfTables; i++) {
  1423.       mysql_stmt_close(prep_insert[i]);
  1424.       mysql_stmt_close(prep_update[i]);
  1425.       mysql_stmt_close(prep_delete[i]);
  1426.       mysql_stmt_close(prep_read[i]);
  1427.     }
  1428.   }
  1429. #endif
  1430.   if (use_ndb && pNdb) {
  1431.     ndbout << "I got here " << endl;
  1432.     return_ndb_object(pNdb, ndb_id);
  1433.   }
  1434.   return NULL;
  1435. }
  1438. static int readArguments(int argc, const char** argv)
  1439. {
  1441.   int i = 1;
  1442.   while (argc > 1){
  1443.     if (strcmp(argv[i], "-t") == 0){
  1444.       tNoOfThreads = atoi(argv[i+1]);
  1445.       if ((tNoOfThreads < 1)) 
  1446.         return -1;
  1447.       argc -= 1;
  1448.       i++;
  1449.     }else if (strcmp(argv[i], "-o") == 0){
  1450.       tNoOfOperations = atoi(argv[i+1]);
  1451.       if (tNoOfOperations < 1) 
  1452.         return -1;;
  1453.       argc -= 1;
  1454.       i++;
  1455.     }else if (strcmp(argv[i], "-a") == 0){
  1456.       tNoOfAttributes = atoi(argv[i+1]);
  1457.       if ((tNoOfAttributes < 2) || (tNoOfAttributes > MAXATTR)) 
  1458.         return -1;
  1459.       argc -= 1;
  1460.       i++;
  1461.     }else if (strcmp(argv[i], "-c") == 0){
  1462.       tNoOfTables = atoi(argv[i+1]);
  1463.       if ((tNoOfTables < 1) || (tNoOfTables > MAXTABLES)) 
  1464.         return -1;
  1465.       argc -= 1;
  1466.       i++;
  1467.     }else if (strcmp(argv[i], "-stdtables") == 0){
  1468.       theStdTableNameFlag = 1;
  1469.     }else if (strcmp(argv[i], "-l") == 0){
  1470.       tNoOfLoops = atoi(argv[i+1]);
  1471.       if ((tNoOfLoops < 0) || (tNoOfLoops > 100000)) 
  1472.         return -1;
  1473.       argc -= 1;
  1474.       i++;
  1475.     }else if (strcmp(argv[i], "-pool_size") == 0){
  1476.       t_instances = atoi(argv[i+1]);
  1477.       if ((t_instances < 1) || (t_instances > 240)) 
  1478.         return -1;
  1479.       argc -= 1;
  1480.       i++;
  1481. #ifdef USE_MYSQL
  1482.     }else if (strcmp(argv[i], "-engine") == 0){
  1483.       engine_id = atoi(argv[i+1]);
  1484.       if ((engine_id < 0) || (engine_id > 3)) 
  1485.         return -1;
  1486.       argc -= 1;
  1487.       i++;
  1488.     }else if (strcmp(argv[i], "-socket") == 0){
  1489.       sockets[n_sockets] = atoi(argv[i+1]);
  1490.       if (sockets[n_sockets] <= 0)
  1491.         return -1;
  1492.       n_sockets++;
  1493.       argc -= 1;
  1494.       i++;
  1495.     }else if (strcmp(argv[i], "-use_ndb") == 0){
  1496.       use_ndb = true;
  1497. #endif
  1498.     }else if (strcmp(argv[i], "-s") == 0){
  1499.       tAttributeSize = atoi(argv[i+1]);
  1500.       if ((tAttributeSize < 1) || (tAttributeSize > MAXATTRSIZE)) 
  1501.         return -1;
  1502.       argc -= 1;
  1503.       i++;
  1504.     }else if (strcmp(argv[i], "-lkn") == 0){
  1505.      tNoOfLongPK = atoi(argv[i+1]);
  1506.      useLongKeys = true;
  1507.       if ((tNoOfLongPK < 1) || (tNoOfLongPK > MAXNOLONGKEY) || 
  1508.   (tNoOfLongPK * tSizeOfLongPK) > MAXLONGKEYTOTALSIZE){
  1509.        ndbout << "Argument -lkn is not in the proper range." << endl;  
  1510. return -1;
  1511.       }
  1512.       argc -= 1;
  1513.       i++;
  1514.     }else if (strcmp(argv[i], "-lks") == 0){
  1515.       tSizeOfLongPK = atoi(argv[i+1]);
  1516.       useLongKeys = true;
  1517.       if ((tSizeOfLongPK < 1) || (tNoOfLongPK * tSizeOfLongPK) > MAXLONGKEYTOTALSIZE){
  1518. ndbout << "Argument -lks is not in the proper range 1 to " << 
  1519.   MAXLONGKEYTOTALSIZE << endl;
  1520.         return -1;
  1521.       }
  1522.       argc -= 1;
  1523.       i++;
  1524.     }else if (strcmp(argv[i], "-simple") == 0){
  1525.       theSimpleFlag = 1;
  1526.     }else if (strcmp(argv[i], "-write") == 0){
  1527.       theWriteFlag = 1;
  1528.     }else if (strcmp(argv[i], "-dirty") == 0){
  1529.       theDirtyFlag = 1;
  1530.     }else if (strcmp(argv[i], "-sleep") == 0){
  1531.       tSleepTime = atoi(argv[i+1]);
  1532.       if ((tSleepTime < 1) || (tSleepTime > 3600)) 
  1533.         return -1;
  1534.       argc -= 1;
  1535.       i++;
  1536.     }else if (strcmp(argv[i], "-no_table_create") == 0){
  1537.       theTableCreateFlag = 1;
  1538.     }else if (strcmp(argv[i], "-temp") == 0){
  1539.       theTempTable = true;
  1540.     }else if (strcmp(argv[i], "-noverify") == 0){
  1541.       VerifyFlag = false ;
  1542.     }else if (theErrorData.parseCmdLineArg(argv, i) == true){
  1543.       ; //empty, updated in errorArg(..)
  1544.     }else if (strcmp(argv[i], "-verify") == 0){
  1545.       VerifyFlag = true ;
  1546. #ifdef CEBIT_STAT
  1547.     }else if (strcmp(argv[i], "-statserv") == 0){
  1548.       if (! (argc > 2))
  1549. return -1;
  1550.       const char *p = argv[i+1];
  1551.       const char *q = strrchr(p, ':');
  1552.       if (q == 0)
  1553. return -1;
  1554.       BaseString::snprintf(statHost, sizeof(statHost), "%.*s", q-p, p);
  1555.       statPort = atoi(q+1);
  1556.       statEnable = true;
  1557.       argc -= 1;
  1558.       i++;
  1559.     }else if (strcmp(argv[i], "-statfreq") == 0){
  1560.       if (! (argc > 2))
  1561. return -1;
  1562.       statFreq = atoi(argv[i+1]);
  1563.       if (statFreq < 1)
  1564. return -1;
  1565.       argc -= 1;
  1566.       i++;
  1567. #endif
  1568.     }else{       
  1569.       return -1;
  1570.     }
  1571.     argc -= 1;
  1572.     i++;
  1573.   }
  1574. #ifdef USE_MYSQL
  1575.   if (n_sockets == 0) {
  1576.     n_sockets = 1;
  1577.     sockets[0] = 3306;
  1578.   }
  1579. #endif
  1580.   return 0;
  1581. }
  1583. static void sleepBeforeStartingTest(int seconds){
  1584.   if (seconds > 0){
  1585.       ndbout << "Sleeping(" <<seconds << ")...";
  1586.       NdbSleep_SecSleep(seconds);
  1587.       ndbout << " done!" << endl;
  1588.     }
  1589. }
  1592. #ifdef USE_MYSQL
  1593. static int
  1594. dropTables(MYSQL* mysqlp){
  1595.   char buf[2048];
  1596.   for(unsigned i = 0; i < tNoOfTables; i++){
  1597.     int pos = 0;
  1598.     ndbout << "Dropping " << tableName[i] << "... ";
  1599.     pos += sprintf(buf+pos, "%s", "DROP TABLE ");
  1600.     pos += sprintf(buf+pos, "%s%s", tableName[i], ";");
  1601.     if (verbose)
  1602.       ndbout << endl << buf << endl;
  1603.     if (mysql_query(mysqlp, buf) != 0){
  1604.       ndbout << "Failed!"<<endl
  1605.      <<mysql_error(mysqlp)<<endl
  1606.      <<buf<<endl;
  1607.     } else
  1608.       ndbout << "OK!" << endl;
  1609.   }
  1610.   
  1611.   return 0;
  1612. }
  1613. #endif
  1615. #ifdef USE_MYSQL
  1616. static int
  1617. createTables(MYSQL* mysqlp){
  1619.   for (Uint32 i = 0; i < tNoOfAttributes; i++){
  1620.     BaseString::snprintf(attrName[i], MAXSTRLEN, "COL%d", i);
  1621.   }
  1623.   // Note! Uses only uppercase letters in table name's
  1624.   // so that we can look at the tables with SQL
  1625.   for (Uint32 i = 0; i < tNoOfTables; i++){
  1626.     if (theStdTableNameFlag == 0){
  1627.       BaseString::snprintf(tableName[i], MAXSTRLEN, "TAB%d_%d", i, 
  1628.        (int)(NdbTick_CurrentMillisecond() / 1000));
  1629.     } else {
  1630.       BaseString::snprintf(tableName[i], MAXSTRLEN, "TAB%d", i);
  1631.     }
  1632.   }
  1633.   
  1634.   char buf[2048];
  1635.   for(unsigned i = 0; i < tNoOfTables; i++){
  1636.     int pos = 0;
  1637.     ndbout << "Creating " << tableName[i] << "... ";
  1638.     
  1639.     pos += sprintf(buf+pos, "%s", "CREATE TABLE ");
  1640.     pos += sprintf(buf+pos, "%s%s", tableName[i], " ");
  1641.     if(useLongKeys){
  1642.       for(Uint32 i = 0; i < tNoOfLongPK; i++) {
  1643.       }
  1644.     } else {
  1645.       pos += sprintf(buf+pos, "%s%s%s",
  1646.      "(", attrName[0], " int unsigned primary key");
  1647.     }
  1648.     for (unsigned j = 1; j < tNoOfAttributes; j++)
  1649.       pos += sprintf(buf+pos, "%s%s%s", ",", attrName[j], " int unsigned");
  1650.     pos += sprintf(buf+pos, "%s%s%s", ")", engine[engine_id], ";");
  1651.     if (verbose)
  1652.       ndbout << endl << buf << endl;
  1653.     if (mysql_query(mysqlp, buf) != 0)
  1654.       return -1;
  1655.     ndbout << "done" << endl;
  1656.   }
  1657.   return 0;
  1658. }
  1659. #endif
  1661. static int
  1662. createTables(Ndb* pMyNdb){
  1664.   for (Uint32 i = 0; i < tNoOfAttributes; i++){
  1665.     BaseString::snprintf(attrName[i], MAXSTRLEN, "COL%d", i);
  1666.   }
  1668.   // Note! Uses only uppercase letters in table name's
  1669.   // so that we can look at the tables with SQL
  1670.   for (Uint32 i = 0; i < tNoOfTables; i++){
  1671.     if (theStdTableNameFlag == 0){
  1672.       BaseString::snprintf(tableName[i], MAXSTRLEN, "TAB%d_%d", i, 
  1673.        (int)(NdbTick_CurrentMillisecond() / 1000));
  1674.     } else {
  1675.       BaseString::snprintf(tableName[i], MAXSTRLEN, "TAB%d", i);
  1676.     }
  1677.   }
  1678.   
  1679.   for(unsigned i = 0; i < tNoOfTables; i++){
  1680.     ndbout << "Creating " << tableName[i] << "... ";
  1681.     
  1682.     NdbDictionary::Table tmpTable(tableName[i]);
  1683.     
  1684.     tmpTable.setStoredTable(!theTempTable);
  1686.     if(useLongKeys){
  1687.       for(Uint32 i = 0; i < tNoOfLongPK; i++) {
  1688. NdbDictionary::Column col(longKeyAttrName[i]);
  1689. col.setType(NdbDictionary::Column::Unsigned);
  1690. col.setLength(tSizeOfLongPK);
  1691. col.setPrimaryKey(true);
  1692. tmpTable.addColumn(col);
  1693.       }
  1694.     } else {
  1695.       NdbDictionary::Column col(attrName[0]);
  1696.       col.setType(NdbDictionary::Column::Unsigned);
  1697.       col.setLength(1);
  1698.       col.setPrimaryKey(true);
  1699.       tmpTable.addColumn(col);
  1700.     }
  1701.     NdbDictionary::Column col;
  1702.     col.setType(NdbDictionary::Column::Unsigned);
  1703.     col.setLength(tAttributeSize);
  1704.     for (unsigned j = 1; j < tNoOfAttributes; j++){
  1705.       col.setName(attrName[j]);
  1706.       tmpTable.addColumn(col);
  1707.     }
  1708.     if(pMyNdb->getDictionary()->createTable(tmpTable) == -1){
  1709.       return -1;
  1710.     }
  1711.     ndbout << "done" << endl;
  1712.   }
  1713.   
  1714.   return 0;
  1715. }
  1717.       
  1718. static void input_error(){
  1719.   ndbout << endl << "Invalid argument!" << endl;
  1720.   ndbout << endl << "Arguments:" << endl;
  1721.   ndbout << "   -t Number of threads to start, default 1" << endl;
  1722.   ndbout << "   -o Number of operations per loop, default 500" << endl;
  1723.   ndbout << "   -l Number of loops to run, default 1, 0=infinite" << endl;
  1724.   ndbout << "   -a Number of attributes, default 25" << endl;
  1725.   ndbout << "   -c Number of tables, default 1" << endl;
  1726.   ndbout << "   -s Size of each attribute, default 1 (Primary Key is always of size 1," << endl;
  1727.   ndbout << "      independent of this value)" << endl;
  1728.   ndbout << "   -lkn Number of long primary keys, default 1" << endl;
  1729.   ndbout << "   -lks Size of each long primary key, default 1" << endl;
  1731.   ndbout << "   -simple Use simple read to read from database" << endl;
  1732.   ndbout << "   -dirty Use dirty read to read from database" << endl;
  1733.   ndbout << "   -write Use writeTuple in insert and update" << endl;
  1734.   ndbout << "   -stdtables Use standard table names" << endl;
  1735.   ndbout << "   -no_table_create Don't create tables in db" << endl;
  1736.   ndbout << "   -sleep Sleep a number of seconds before running the test, this" << endl;
  1737.   ndbout << "    can be used so that another flexBench have time to create tables" << endl;
  1738.   ndbout << "   -temp Use tables without logging" << endl;
  1739.   ndbout << "   -verify Verify inserts, updates and deletes" << endl ;
  1740.   ndbout << "   -use_ndb Use NDB API (otherwise use mysql client)" << endl ;
  1741.   ndbout << "   -pool_size Number of Ndb objects in pool" << endl ;
  1742.   theErrorData.printCmdLineArgs(ndbout);
  1743.   ndbout << endl <<"Returns:" << endl;
  1744.   ndbout << "t 0 - Test passed" << endl;
  1745.   ndbout << "t 1 - Test failed" << endl;
  1746.   ndbout << "t 2 - Invalid arguments" << endl << endl;
  1747. }
  1749. // vim: set sw=2: