mariadb/ndb/test/ndbapi/mainAsyncGenerator.cpp
2004-09-15 11:49:18 +02:00

393 lines
10 KiB
C++

/* Copyright (C) 2003 MySQL AB
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#include <ndb_global.h>
#include <NdbHost.h>
#include <NdbSleep.h>
#include <NdbThread.h>
#include <NdbMain.h>
#include <NdbOut.hpp>
#include <NdbEnv.h>
#include <NdbTest.hpp>
#include "userInterface.h"
#include "dbGenerator.h"
static int numProcesses;
static int numSeconds;
static int numWarmSeconds;
static int parallellism;
static int millisSendPoll;
static int minEventSendPoll;
static int forceSendPoll;
static ThreadData *data;
static void usage(const char *prog)
{
const char *progname;
/*--------------------------------------------*/
/* Get the name of the program (without path) */
/*--------------------------------------------*/
progname = strrchr(prog, '/');
if (progname == 0)
progname = prog;
else
++progname;
ndbout_c(
"Usage: %s [-proc <num>] [-warm <num>] [-time <num>] [ -p <num>] "
"[-t <num> ] [ -e <num> ] [ -f <num>] \n"
" -proc <num> Specifies that <num> is the number of\n"
" threads. The default is 1.\n"
" -time <num> Specifies that the test will run for <num> sec.\n"
" The default is 10 sec\n"
" -warm <num> Specifies the warm-up/cooldown period of <num> "
"sec.\n"
" The default is 10 sec\n"
" -p <num> The no of parallell transactions started by "
"one thread\n"
" -e <num> Minimum no of events before wake up in call to "
"sendPoll\n"
" Default is 1\n"
" -f <num> force parameter to sendPoll\n"
" Default is 0\n",
progname);
}
static
int
parse_args(int argc, const char **argv)
{
int i;
numProcesses = 1;
numSeconds = 10;
numWarmSeconds = 10;
parallellism = 1;
millisSendPoll = 10000;
minEventSendPoll = 1;
forceSendPoll = 0;
i = 1;
while (i < argc){
if (strcmp("-proc",argv[i]) == 0) {
if (i + 1 >= argc) {
return 1;
}
if (sscanf(argv[i+1], "%d", &numProcesses) == -1 ||
numProcesses <= 0 || numProcesses > 127) {
ndbout_c("-proc flag requires a positive integer argument [1..127]");
return 1;
}
i += 2;
} else if (strcmp("-p", argv[i]) == 0){
if(i + 1 >= argc){
usage(argv[0]);
return 1;
}
if (sscanf(argv[i+1], "%d", &parallellism) == -1 ||
parallellism <= 0){
ndbout_c("-p flag requires a positive integer argument");
return 1;
}
i += 2;
}
else if (strcmp("-time",argv[i]) == 0) {
if (i + 1 >= argc) {
return 1;
}
if (sscanf(argv[i+1], "%d", &numSeconds) == -1 ||
numSeconds < 0) {
ndbout_c("-time flag requires a positive integer argument");
return 1;
}
i += 2;
}
else if (strcmp("-warm",argv[i]) == 0) {
if (i + 1 >= argc) {
return 1;
}
if (sscanf(argv[i+1], "%d", &numWarmSeconds) == -1 ||
numWarmSeconds < 0) {
ndbout_c("-warm flag requires a positive integer argument");
return 1;
}
i += 2;
}
else if (strcmp("-e",argv[i]) == 0) {
if (i + 1 >= argc) {
return 1;
}
if (sscanf(argv[i+1], "%d", &minEventSendPoll) == -1 ||
minEventSendPoll < 0) {
ndbout_c("-e flag requires a positive integer argument");
return 1;
}
i += 2;
}
else if (strcmp("-f",argv[i]) == 0) {
if (i + 1 >= argc) {
usage(argv[0]);
return 1;
}
if (sscanf(argv[i+1], "%d", &forceSendPoll) == -1 ||
forceSendPoll < 0) {
ndbout_c("-f flag requires a positive integer argument");
return 1;
}
i += 2;
}
else {
return 1;
}
}
if(minEventSendPoll > parallellism){
ndbout_c("minEventSendPoll(%d) > parallellism(%d)",
minEventSendPoll, parallellism);
ndbout_c("not very good...");
ndbout_c("very bad...");
ndbout_c("exiting...");
return 1;
}
return 0;
}
static
void
print_transaction(const char *header,
unsigned long totalCount,
TransactionDefinition *trans,
unsigned int printBranch,
unsigned int printRollback)
{
double f;
ndbout_c(" %s: %d (%.2f%%) "
"Latency(ms) avg: %d min: %d max: %d std: %d n: %d",
header,
trans->count,
(double)trans->count / (double)totalCount * 100.0,
(int)trans->latency.getMean(),
(int)trans->latency.getMin(),
(int)trans->latency.getMax(),
(int)trans->latency.getStddev(),
(int)trans->latency.getCount()
);
if( printBranch ){
if( trans->count == 0 )
f = 0.0;
else
f = (double)trans->branchExecuted / (double)trans->count * 100.0;
ndbout_c(" Branches Executed: %d (%.2f%%)", trans->branchExecuted, f);
}
if( printRollback ){
if( trans->count == 0 )
f = 0.0;
else
f = (double)trans->rollbackExecuted / (double)trans->count * 100.0;
ndbout_c(" Rollback Executed: %d (%.2f%%)",trans->rollbackExecuted,f);
}
}
void
print_stats(const char *title,
unsigned int length,
unsigned int transactionFlag,
GeneratorStatistics *gen,
int numProc, int parallellism)
{
int i;
char buf[10];
char name[MAXHOSTNAMELEN];
name[0] = 0;
NdbHost_GetHostName(name);
ndbout_c("\n------ %s ------",title);
ndbout_c("Length : %d %s",
length,
transactionFlag ? "Transactions" : "sec");
ndbout_c("Processor : %s", name);
ndbout_c("Number of Proc: %d",numProc);
ndbout_c("Parallellism : %d", parallellism);
ndbout_c("\n");
if( gen->totalTransactions == 0 ) {
ndbout_c(" No Transactions for this test");
}
else {
for(i = 0; i < 5; i++) {
sprintf(buf, "T%d",i+1);
print_transaction(buf,
gen->totalTransactions,
&gen->transactions[i],
i >= 2,
i >= 3 );
}
ndbout_c("\n");
ndbout_c(" Overall Statistics:");
ndbout_c(" Transactions: %d", gen->totalTransactions);
ndbout_c(" Outer : %.0f TPS",gen->outerTps);
ndbout_c("\n");
}
}
static
void *
threadRoutine(void *arg)
{
int i;
ThreadData *data = (ThreadData *)arg;
Ndb * pNDB;
pNDB = asyncDbConnect(parallellism);
/* NdbSleep_MilliSleep(rand() % 10); */
for(i = 0; i<parallellism; i++){
data[i].pNDB = pNDB;
}
millisSendPoll = 30000;
asyncGenerator(data, parallellism,
millisSendPoll, minEventSendPoll, forceSendPoll);
asyncDbDisconnect(pNDB);
NdbThread_Exit(0);
return NULL;
}
NDB_COMMAND(DbAsyncGenerator, "DbAsyncGenerator",
"DbAsyncGenerator", "DbAsyncGenerator", 65535)
{
ndb_init();
int i;
int j;
int k;
struct NdbThread* pThread = NULL;
GeneratorStatistics stats;
GeneratorStatistics *p;
char threadName[32];
int rc = NDBT_OK;
void* tmp = NULL;
if(parse_args(argc,argv) != 0){
usage(argv[0]);
return NDBT_ProgramExit(NDBT_WRONGARGS);
}
ndbout_c("\nStarting Test with %d process(es) for %d %s parallellism %d",
numProcesses,
numSeconds,
"sec",
parallellism);
ndbout_c(" WarmUp/coolDown = %d sec", numWarmSeconds);
data = (ThreadData*)malloc((numProcesses*parallellism)*sizeof(ThreadData));
for(i = 0; i < numProcesses; i++) {
for(j = 0; j<parallellism; j++){
data[i*parallellism+j].warmUpSeconds = numWarmSeconds;
data[i*parallellism+j].testSeconds = numSeconds;
data[i*parallellism+j].coolDownSeconds = numWarmSeconds;
data[i*parallellism+j].randomSeed =
NdbTick_CurrentMillisecond()+i+j;
data[i*parallellism+j].changedTime = 0;
data[i*parallellism+j].runState = Runnable;
}
sprintf(threadName, "AsyncThread[%d]", i);
pThread = NdbThread_Create(threadRoutine,
(void**)&data[i*parallellism],
65535,
threadName,
NDB_THREAD_PRIO_LOW);
if(pThread != 0 && pThread != NULL){
(&data[i*parallellism])->pThread = pThread;
} else {
perror("Failed to create thread");
rc = NDBT_FAILED;
}
}
showTime();
/*--------------------------------*/
/* Wait for all processes to exit */
/*--------------------------------*/
for(i = 0; i < numProcesses; i++) {
NdbThread_WaitFor(data[i*parallellism].pThread, &tmp);
NdbThread_Destroy(&data[i*parallellism].pThread);
}
ndbout_c("All threads have finished");
/*-------------------------------------------*/
/* Clear all structures for total statistics */
/*-------------------------------------------*/
stats.totalTransactions = 0;
stats.outerTps = 0.0;
for(i = 0; i < NUM_TRANSACTION_TYPES; i++ ) {
stats.transactions[i].count = 0;
stats.transactions[i].branchExecuted = 0;
stats.transactions[i].rollbackExecuted = 0;
stats.transactions[i].latency.reset();
}
/*--------------------------------*/
/* Add the values for all Threads */
/*--------------------------------*/
for(i = 0; i < numProcesses; i++) {
for(k = 0; k<parallellism; k++){
p = &data[i*parallellism+k].generator;
stats.totalTransactions += p->totalTransactions;
stats.outerTps += p->outerTps;
for(j = 0; j < NUM_TRANSACTION_TYPES; j++ ) {
stats.transactions[j].count +=
p->transactions[j].count;
stats.transactions[j].branchExecuted +=
p->transactions[j].branchExecuted;
stats.transactions[j].rollbackExecuted +=
p->transactions[j].rollbackExecuted;
stats.transactions[j].latency +=
p->transactions[j].latency;
}
}
}
print_stats("Test Results",
numSeconds,
0,
&stats,
numProcesses,
parallellism);
free(data);
NDBT_ProgramExit(rc);
}