msa.cpp

MySQL数据库开发源码 值得一看哦

/* 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 <NdbApi.hpp>
#include <NdbSchemaCon.hpp>
#include <NdbCondition.h>
#include <NdbMutex.h>
#include <NdbSleep.h>
#include <NdbThread.h>
#include <NdbTick.h>

const char* const c_szDatabaseName = "TEST_DB";

const char* const c_szTableNameStored = "CCStored";
const char* const c_szTableNameTemp = "CCTemp";

const char* const c_szContextId = "ContextId";
const char* const c_szVersion = "Version";
const char* const c_szLockFlag = "LockFlag";
const char* const c_szLockTime = "LockTime";
const char* const c_szLockTimeUSec = "LockTimeUSec";
const char* const c_szContextData = "ContextData";

const char* g_szTableName = c_szTableNameStored;


#ifdef NDB_WIN32
HANDLE hShutdownEvent = 0;
#else
#include <signal.h>
bool bShutdownEvent = false;
#endif
long g_nMaxContextIdPerThread = 5000;
long g_nNumThreads = 0;
long g_nMaxCallsPerSecond = 0;
long g_nMaxRetry = 50;
bool g_bWriteTuple = false;
bool g_bInsertInitial = false;
bool g_bVerifyInitial = false;

NdbMutex* g_pNdbMutexPrintf = 0;
NdbMutex* g_pNdbMutexIncrement = 0;
long g_nNumCallsProcessed = 0;
NDB_TICKS g_tStartTime = 0;
NDB_TICKS g_tEndTime = 0;

long g_nNumberOfInitialInsert = 0;
long g_nNumberOfInitialVerify = 0;

const long c_nMaxMillisecForAllCall = 5000;
long* g_plCountMillisecForCall = 0;
const long c_nMaxMillisecForAllTrans = 5000;
long* g_plCountMillisecForTrans = 0;
bool g_bReport = false;
bool g_bReportPlus = false;


// data for CALL_CONTEXT and GROUP_RESOURCE
static char STATUS_DATA[]= 
"000102030405060708090A0B0C0D0E0F000102030405060708090A0B0C0D0E0F"
"101112131415161718191A1B1C1D1E1F000102030405060708090A0B0C0D0E0F"
"202122232425262728292A2B2C2D2E2F000102030405060708090A0B0C0D0E0F"
"303132333435363738393A3B3C3D3E3F000102030405060708090A0B0C0D0E0F"
"404142434445464748494A4B4C4D4E4F000102030405060708090A0B0C0D0E0F"
"505152535455565758595A5B5C5D5E5F000102030405060708090A0B0C0D0E0F"
"606162636465666768696A6B6C6D6E6F000102030405060708090A0B0C0D0E0F"
"707172737475767778797A7B7C7D7E7F000102030405060708090A0B0C0D0E0F"
"808182838485868788898A8B8C8D8E8F000102030405060708090A0B0C0D0E0F"
"909192939495969798999A9B9C9D9E9F000102030405060708090A0B0C0D0E0F"
"10010110210310410510610710810910A000102030405060708090A0B0C0D0EF"
"10B10C10D10E10F110111112113114115000102030405060708090A0B0C0D0EF"
"11611711811911A11B11C11D11E11F120000102030405060708090A0B0C0D0EF"
"12112212312412512612712812912A12B000102030405060708090A0B0C0D0EF"
"12C12D12E12F130131132134135136137000102030405060708090A0B0C0D0EF"
"13813913A13B13C13D13E13F140141142000102030405060708090A0B0C0D0EF"
"14314414514614714814914A14B14C14D000102030405060708090A0B0C0D0EF"
"14E14F150151152153154155156157158000102030405060708090A0B0C0D0EF"
"15915A15B15C15D15E15F160161162163000102030405060708090A0B0C0D0EF"
"16416516616716816916A16B16C16D16E000102030405060708090A0B0C0D0EF"
"16F170171172173174175176177178179000102030405060708090A0B0C0D0EF"
"17A17B17C17D17E17F180181182183184000102030405060708090A0B0C0D0EF"
"18518618718818918A18B18C18D18E18F000102030405060708090A0B0C0D0EF"
"19019119219319419519619719819919A000102030405060708090A0B0C0D0EF"
"19B19C19D19E19F200201202203204205000102030405060708090A0B0C0D0EF"
"20620720820920A20B20C20D20F210211000102030405060708090A0B0C0D0EF"
"21221321421521621721821921A21B21C000102030405060708090A0B0C0D0EF"
"21D21E21F220221222223224225226227000102030405060708090A0B0C0D0EF"
"22822922A22B22C22D22E22F230231232000102030405060708090A0B0C0D0EF"
"23323423523623723823923A23B23C23D000102030405060708090A0B0C0D0EF"
"23E23F240241242243244245246247248000102030405060708090A0B0C0D0EF"
"24924A24B24C24D24E24F250251252253000102030405060708090A0B0C0D0EF"
"101112131415161718191A1B1C1D1E1F000102030405060708090A0B0C0D0E0F"
"202122232425262728292A2B2C2D2E2F000102030405060708090A0B0C0D0E0F"
"303132333435363738393A3B3C3D3E3F000102030405060708090A0B0C0D0E0F"
"404142434445464748494A4B4C4D4E4F000102030405060708090A0B0C0D0E0F"
"505152535455565758595A5B5C5D5E5F000102030405060708090A0B0C0D0E0F"
"606162636465666768696A6B6C6D6E6F000102030405060708090A0B0C0D0E0F"
"707172737475767778797A7B7C7D7E7F000102030405060708090A0B0C0D0E0F"
"808182838485868788898A8B8C8D8E8F000102030405060708090A0B0C0D0E0F"
"909192939495969798999A9B9C9D9E9F000102030405060708090A0B0C0D0E0F"
"10010110210310410510610710810910A000102030405060708090A0B0C0D0EF"
"10B10C10D10E10F110111112113114115000102030405060708090A0B0C0D0EF"
"11611711811911A11B11C11D11E11F120000102030405060708090A0B0C0D0EF"
"12112212312412512612712812912A12B000102030405060708090A0B0C0D0EF"
"12C12D12E12F130131132134135136137000102030405060708090A0B0C0D0EF"
"13813913A13B13C13D13E13F140141142000102030405060708090A0B0C0D0EF"
"14314414514614714814914A14B14C14D000102030405060708090A0B0C0D0EF"
"14E14F150151152153154155156157158000102030405060708090A0B0C0D0EF"
"15915A15B15C15D15E15F160161162163000102030405060708090A0B0C0D0EF"
"16416516616716816916A16B16C16D16E000102030405060708090A0B0C0D0EF"
"16F170171172173174175176177178179000102030405060708090A0B0C0D0EF"
"17A17B17C17D17E17F180181182183184000102030405060708090A0B0C0D0EF"
"18518618718818918A18B18C18D18E18F000102030405060708090A0B0C0D0EF"
"19019119219319419519619719819919A000102030405060708090A0B0C0D0EF"
"19B19C19D19E19F200201202203204205000102030405060708090A0B0C0D0EF"
"20620720820920A20B20C20D20F210211000102030405060708090A0B0C0D0EF"
"21221321421521621721821921A21B21C000102030405060708090A0B0C0D0EF"
"21D21E21F220221222223224225226227000102030405060708090A0B0C0D0EF"
"22822922A22B22C22D22E22F230231232000102030405060708090A0B0C0D0EF"
"23323423523623723823923A23B23C23D000102030405060708090A0B0C0D0EF"
"2366890FE1438751097E7F6325DC0E6326F"
"25425525625725825925A25B25C25D25E25F000102030405060708090A0B0C0F";     

long g_nStatusDataSize = sizeof(STATUS_DATA);


// Thread function for Call Context Inserts


#ifdef NDB_WIN32

BOOL WINAPI ConsoleCtrlHandler(DWORD dwCtrlType)
{
    if(CTRL_C_EVENT == dwCtrlType)
    {
        SetEvent(hShutdownEvent);
        return TRUE;
    }
    return FALSE;
}

#else

void CtrlCHandler(int)
{
    bShutdownEvent = true;
}

#endif



void ReportNdbError(const char* szMsg, const NdbError& err)
{
    NdbMutex_Lock(g_pNdbMutexPrintf);
    printf("%s: %d: %s\n", szMsg, err.code, (err.message ? err.message : ""));
    NdbMutex_Unlock(g_pNdbMutexPrintf);
}


void
ReportCallsPerSecond(long nNumCallsProcessed, 
                     NDB_TICKS tStartTime, 
                     NDB_TICKS tEndTime)
{
    NDB_TICKS tElapsed = tEndTime - tStartTime;
    long lCallsPerSec;
    if(tElapsed>0)
        lCallsPerSec = (long)((1000*nNumCallsProcessed)/tElapsed);
    else
        lCallsPerSec = 0;

    NdbMutex_Lock(g_pNdbMutexPrintf);
    printf("Time Taken for %ld Calls is %ld msec (= %ld calls/sec)\n",
        nNumCallsProcessed, (long)tElapsed, lCallsPerSec);
    NdbMutex_Unlock(g_pNdbMutexPrintf);
}


#ifndef NDB_WIN32
void InterlockedIncrement(long* lp)             // expensive
{
    NdbMutex_Lock(g_pNdbMutexIncrement);
    (*lp)++;
    NdbMutex_Unlock(g_pNdbMutexIncrement);
}
#endif


void InterlockedIncrementAndReport(void)
{
    NdbMutex_Lock(g_pNdbMutexIncrement);
    ++g_nNumCallsProcessed;
    if((g_nNumCallsProcessed%1000)==0) 
    {
        g_tEndTime = NdbTick_CurrentMillisecond();
        if(g_tStartTime) 
            ReportCallsPerSecond(1000, g_tStartTime, g_tEndTime);

        g_tStartTime = g_tEndTime;
    }
    NdbMutex_Unlock(g_pNdbMutexIncrement);
}


void SleepOneCall(void)
{
    int iMillisecToSleep;
    if(g_nMaxCallsPerSecond>0)
        iMillisecToSleep = (1000*g_nNumThreads)/g_nMaxCallsPerSecond;
    else
        iMillisecToSleep = 50;

    if(iMillisecToSleep>0)
        NdbSleep_MilliSleep(iMillisecToSleep);

}



int QueryTransaction(Ndb* pNdb, 
                     long iContextId,                       
                     long* piVersion, 
                     long* piLockFlag, 
                     long* piLockTime, 
                     long* piLockTimeUSec, 
                     char* pchContextData, 
                     NdbError& err)
{
    int iRes = -1;
    NdbConnection* pNdbConnection = pNdb->startTransaction(0, (const char*)&iContextId, 4);
    if(pNdbConnection)
    {
        NdbOperation* pNdbOperation = pNdbConnection->getNdbOperation(g_szTableName);
        if(pNdbOperation)
        {
            NdbRecAttr* pNdbRecAttrVersion;
            NdbRecAttr* pNdbRecAttrLockFlag;
            NdbRecAttr* pNdbRecAttrLockTime;
            NdbRecAttr* pNdbRecAttrLockTimeUSec;
            NdbRecAttr* pNdbRecAttrContextData;
            if(!pNdbOperation->readTuple()
            && !pNdbOperation->equal(c_szContextId, (Int32)iContextId)
            && (pNdbRecAttrVersion=pNdbOperation->getValue(c_szVersion, (char*)piVersion))
            && (pNdbRecAttrLockFlag=pNdbOperation->getValue(c_szLockFlag, (char*)piLockFlag))
            && (pNdbRecAttrLockTime=pNdbOperation->getValue(c_szLockTime, (char*)piLockTime))
            && (pNdbRecAttrLockTimeUSec=pNdbOperation->getValue(c_szLockTimeUSec, (char*)piLockTimeUSec))
            && (pNdbRecAttrContextData=pNdbOperation->getValue(c_szContextData, pchContextData)))
            {
                if(!pNdbConnection->execute(Commit))
                    iRes = 0;
                else 
                    err = pNdbConnection->getNdbError();
            } 
            else 
                err = pNdbOperation->getNdbError();
        } 
        else 
            err = pNdbConnection->getNdbError();

        pNdb->closeTransaction(pNdbConnection);
    } 
    else 
        err = pNdb->getNdbError();
    
    return iRes;
}


int RetryQueryTransaction(Ndb* pNdb, 
                          long iContextId, 
                          long* piVersion, 
                          long* piLockFlag, 
                          long* piLockTime, 
                          long* piLockTimeUSec, 
                          char* pchContextData, 
                          NdbError& err, 
                          int& nRetry)
{
    int iRes = -1;
    nRetry = 0;
    bool bRetry = true;
    while(bRetry && nRetry<g_nMaxRetry)
    {
        if(!QueryTransaction(pNdb, iContextId, piVersion, piLockFlag, 
            piLockTime, piLockTimeUSec, pchContextData, err))
        {
            iRes = 0;
            bRetry = false;
        }
        else
        {
            switch(err.status)
            {
            case NdbError::TemporaryError:
            case NdbError::UnknownResult:
                SleepOneCall();
                ++nRetry;
                break;
            
            case NdbError::PermanentError:
            default:
                bRetry = false;
                break;
            }
        }
    }
    return iRes;
}


int DeleteTransaction(Ndb* pNdb, long iContextId, NdbError& err)
{
    int iRes = -1;
    NdbConnection* pNdbConnection = pNdb->startTransaction(0, (const char*)&iContextId, 4);
    if(pNdbConnection)
    {
        NdbOperation* pNdbOperation = pNdbConnection->getNdbOperation(g_szTableName);
        if(pNdbOperation)
        {
            if(!pNdbOperation->deleteTuple()
            && !pNdbOperation->equal(c_szContextId, (Int32)iContextId)) 
            {
                if(pNdbConnection->execute(Commit) == 0) 
                    iRes = 0;
                else 
                    err = pNdbConnection->getNdbError();
            } 
            else 
                err = pNdbOperation->getNdbError();
        } 
        else 
            err = pNdbConnection->getNdbError();

        pNdb->closeTransaction(pNdbConnection);
    } 
    else 
        err = pNdb->getNdbError();

    return iRes;
}



int RetryDeleteTransaction(Ndb* pNdb, long iContextId, NdbError& err, int& nRetry)
{
    int iRes = -1;
    nRetry = 0;
    bool bRetry = true;
    bool bUnknown = false;
    while(bRetry && nRetry<g_nMaxRetry)
    {
        if(!DeleteTransaction(pNdb, iContextId, err))
        {
            iRes = 0;
            bRetry = false;
        }
        else
        {
            switch(err.status)
            {
            case NdbError::UnknownResult:
                bUnknown = true;
                ++nRetry;
                break;

            case NdbError::TemporaryError:
                bUnknown = false;
                SleepOneCall();
                ++nRetry;
                break;
            
            case NdbError::PermanentError:
                if(err.code==626 && bUnknown)
                    iRes = 0;
                bRetry = false;
                break;

            default:
                bRetry = false;
                break;
            }
        }
    }
    return iRes;
}