tpc.cpp

全新的纯内存式实时数据库

/*******************************************************************
 *                                                                 *
 *  tpc.cpp                                                      *
 *                                                                 *
 *  This file is a part of the eXtremeDB source code               *
 *  Copyright (c) 2001-2007 McObject LLC                           * 
 *  All Rights Reserved                                            *
 *                                                                 *
 *******************************************************************/
/* 
 *  This is a sample implementation of the Transaction Processing Performance 
 *  Council Benchmark B coded in Java and ANSI SQL2. 
 */ 

#include "mcosql.h"
#include "tpcdb.h"
#include "platform.h"
#ifndef _WIN32
#include "pthread.h"
#endif

using namespace McoSql;

/* tpc bm b scaling rules */
const int tps       =       1;  /* the tps scaling factor: here it is 1 */
const int nbranches =       1;  /* number of branches in 1 tps db       */
const int ntellers  =      10;  /* number of tellers in  1 tps db       */
const int naccounts =  100000;  /* number of accounts in 1 tps db       */
    
const int TELLER  = 0;
const int BRANCH  = 1;
const int ACCOUNT = 2;

#ifndef MCO_PLATFORM_X64
size_t const PAGE_SIZE = 128; // define eXtremeDB page size
#else
size_t const PAGE_SIZE = 256;
#endif

size_t const DATABASE_SIZE = 256*1024*1024;
void*  const MAP_ADDRESS = (void*)0x20000000;
   
McoMultithreadedSqlEngine* engine;  

volatile int n_clients = 10;

int    transaction_count = 0;
int    n_txn_per_client = 10000;
time_t start_time = 0;    
size_t min_memory = DATABASE_SIZE;
size_t max_memory = 0;
bool   initialize_dataset = true;
    
MUTEX_T cs;

#if 1

void reportDone()
{
    MUTEX_LOCK(&cs);
    n_clients--;
    if (n_clients <= 0) {
        time_t end_time = msec(); 
        time_t completion_time = end_time - start_time;
        printf("* Benchmark finished *\n");
        printf("\n* Benchmark Report *" );
        printf("--------------------\n\n");
        printf("Time to execute %d transactions: %ld microseconds.\n", transaction_count, completion_time);
        printf("Max/Min memory usage: %ld/%ld bytes\n",(long) max_memory, (long)min_memory);
        printf("Transaction rate: %f txn/msec.\n", (double)transaction_count / completion_time);
    }    
    MUTEX_UNLOCK(&cs);
}
    
void incrementTransactionCount()
{
    MUTEX_LOCK(&cs);
    transaction_count++;
    MUTEX_UNLOCK(&cs);
}
    
             
char* makeString(char ch, int n) { 
    char* arr = new char[n+1];
    arr[n] = '\0';
    while (--n >= 0) { 
        arr[n] = ch;
    }
    return arr;
}

char const* branchesFiller = makeString(' ', (100 - 16) / 2);
char const* tellersFiller  = makeString(' ', (100 - 20) / 2);
char const* accountsFiller = makeString(' ', (100 - 20) / 2);
char const* historyFiller  = makeString(' ', (50 - 32) / 2);

/*
 *  createDatabase() - Creates and Initializes a scaled database.     
 */ 
void createDatabase() 
{
    int i;
    /* prime database using TPC BM B scaling rules.  
     *  Note that for each branch and teller:
     *      branch_id = teller_id  / ntellers
     *      branch_id = account_id / naccounts
     */    
    for (i = 0; i < nbranches * tps; i++) {
        engine->executeStatement("INSERT INTO branches(Bid,Bbalance,filler) VALUES (%i,0,%s)", i, branchesFiller);
    }
    for (i = 0; i < ntellers * tps; i++) {
        engine->executeStatement("INSERT INTO tellers(Tid,Bid,Tbalance,filler) VALUES (%i,%i,0,%s)", 
                                i, i / ntellers, tellersFiller);
    }
    for ( i = 0; i < naccounts*tps; i++) {
        engine->executeStatement("INSERT INTO accounts(Aid,Bid,Abalance,filler) VALUES (%i,%i,0,%s)", 
                                i, i / naccounts, accountsFiller);
    }
} /* end of CreateDatabase    */
    
                   
int getRandomInt(int lo, int hi)
{
    int ret = 0;
    
    ret = (int)((double)rand() / RAND_MAX * (hi - lo + 1));
    ret += lo;
    
    return ret;
}
    
int getRandomID(int type)
{
    int min, max, num;
    
    max = min = 0;
    num = naccounts;
    
    switch(type) {
      case TELLER:
        min += nbranches;
        num = ntellers;
        /* FALLTHROUGH */
      case BRANCH:
        if (type == BRANCH) { 
            num = nbranches;
        }
        min += naccounts;
        /* FALLTHROUGH */
      case ACCOUNT:
        max = min + num - 1;
    }
    return (getRandomInt(min, max));
}
 
/*
 *  doOne() - Executes a single TPC BM B transaction.
 */

int doOne(int bid, int tid, int aid, int delta)
{
    McoSqlSession session(engine);
    PreparedStatement stmt;
    session.prepare(stmt, "UPDATE accounts SET Abalance=Abalance+%*i WHERE Aid=%*i", &delta, &aid);
    DataSource* result = session.executeQuery("SELECT Abalance FROM  accounts WHERE Aid=%i", aid);
    Cursor* cursor = result->records();
    int aBalance = 0;
                
    while (cursor->hasNext()) {
        aBalance = (int)cursor->next()->get(0)->intValue();
    }
    result->release();

    session.executeStatement("UPDATE tellers SET Tbalance=Tbalance+%i WHERE Tid=%i", delta, tid);
    session.executeStatement("UPDATE branches SET Bbalance=Bbalance+%i WHERE Bid=%i", delta, bid);
                
    session.executeStatement("INSERT INTO account_history(Tid, Bid, Aid, delta, htime, filler) VALUES (%i,%i,%i,%i,now,%s)",
                            tid, bid, aid, delta, historyFiller);
#ifdef _VXWORKS   
    Sleep(0);
#endif //_VXWORKS   

    return aBalance;
}

THREAD_PROC_DEFINE(clientThread, param) 
{
    McoSqlSession session(engine);
    int bid, tid, aid, delta;
    PreparedStatement stmt[5];
    session.prepare(stmt[0], "UPDATE accounts SET Abalance=Abalance+%*i WHERE Aid=%*i", &delta, &aid);
    session.prepare(stmt[1], "SELECT Abalance FROM  accounts WHERE Aid=%*i", &aid);
    session.prepare(stmt[2], "UPDATE tellers SET Tbalance=Tbalance+%*i WHERE Tid=%*i", &delta, &tid);
    session.prepare(stmt[3], "UPDATE branches SET Bbalance=Bbalance+%*i WHERE Bid=%*i", &delta, &bid);
    session.prepare(stmt[4], "INSERT INTO account_history(Tid, Bid, Aid, delta, htime, filler) VALUES (%*i,%*i,%*i,%*i,now,%s)",
                    &tid, &bid, &aid, &delta, historyFiller);

    for (int i = n_txn_per_client; --i >= 0;) {                
        aid = getRandomID(ACCOUNT);
        bid = getRandomID(BRANCH);
        tid = getRandomID(TELLER);
        delta = getRandomInt(0,1000);
        
        session.executePreparedStatement(stmt[0]);
        DataSource* result = session.executePreparedQuery(stmt[1]);
        Cursor* cursor = result->records();
        int aBalance = 0;
        
        while (cursor->hasNext()) {
            aBalance = (int)cursor->next()->get(0)->intValue();
        }
        result->release();

        session.executePreparedStatement(stmt[2]);
        session.executePreparedStatement(stmt[3]);
        session.executePreparedStatement(stmt[4]);

        incrementTransactionCount();
    }
    reportDone();
}
                       

void startBenchmark()
{
    if (initialize_dataset) {
        printf("Initializing dataset...\n");
        createDatabase();
        printf("done.\n\n");
    }
    printf("* Starting Benchmark Run *\n");
            
    start_time = msec();
    THREAD_ID* threads = new THREAD_ID[n_clients];
    for (int i = 0; i < n_clients; i++) {
        THREAD_PROC_START(clientThread, NULL, &threads[i]);
    }

    while (true) { 
        bool stop = false;
        MUTEX_LOCK(&cs);
        stop = (n_clients == 0); 
        MUTEX_UNLOCK(&cs);
        if (stop) { 
            break;
        }

        size_t used = ((MultithreadedAllocator*)MemoryManager::allocator)->usedMemory();
        if (used < min_memory) { 
            min_memory = used;
        } else if (used > max_memory) { 
            max_memory = used;
        }

        Sleep(100);
    }
}

#endif

int main(int argc, char* argv[])
{
    for (int i = 1; i < argc; i++) {
        if (strcmp(argv[i], "-clients") == 0) {
            if (i + 1 < argc) {
                i++;
                n_clients = atoi(argv[i]);
            }
        }
        else if (strcmp(argv[i], "-tpc") == 0) {
            if (i + 1 < argc) {
                i++;
                n_txn_per_client = atoi(argv[i]);
            }
        }
        else if (strcmp(argv[i], "-init") == 0) {
            initialize_dataset = true;
        }
        else {             
            printf("usage: tpc [-init] [-tpc n] [-clients n]\n");
            printf("\t-init\tinitialize the tables\n");
            printf("\t-tpc\ttransactions per client\n");
            printf("\t-clients\tnumber of simultaneous clients\n"); 
            return 1;
        }
    }  
    printf("*********************************************************\n");
    printf("* TPC-B Bench v1.0                                      *\n");
    printf("*********************************************************\n");
    printf("Number of clients: %d\n", n_clients);
    printf("Number of transactions per client: %d\n", n_txn_per_client);

    engine = new McoMultithreadedSqlEngine();
    engine->open("tpcdb", tpcdb_get_dictionary(), DATABASE_SIZE, PAGE_SIZE, MAP_ADDRESS);
    MUTEX_INIT(&cs);
    
    startBenchmark();

    puts("Press any key to exit");
    getchar();

    engine->close();
    MUTEX_DESTROY(&cs);

    return 0;
}