insertrecs.cpp

来自「MySQL数据库开发源码 值得一看哦」· C++ 代码 · 共 572 行 · 第 1/2 页

                        {
                            throw pNdbConnection;
                        }
                        if(pNdbOperation->updateTuple())
                        {
                            throw pNdbOperation;
                        }
                        if(pNdbOperation->equal(_T("ContextId"), nStartingRecordID)
                            || pNdbOperation->setValue(_T("ContextData"), STATUS_DATA, sizeof(STATUS_DATA)))
                        {
                            throw pNdbOperation;
                        }
                        if(pNdbConnection->execute(Commit))
                        {
                            throw pNdbConnection;
                        }
                        pNdb->closeTransaction(pNdbConnection);

						if (bTimeLatency)
						{
							QueryPerformanceCounter(&liEndTime);
							printf("Update = %d msec.\n", (liEndTime.QuadPart - liStartTime.QuadPart) / (freq.QuadPart/1000));
						}
                        
						break;
                }
            }

			nNumCallsProcessed++;

            InterlockedIncrement(pData->pnNumCallsProcessed);
        }

        delete pNdb;
    }
    catch(Ndb* pNdb)
    {
        printf("%d: \n\t%s\n\t%s\n",
            pNdb->getNdbError(), 
            pNdb->getNdbErrorString(), 
            "Ndb");
        delete pNdb;
    }
    catch(NdbConnection* pNdbConnection)
    {
        printf("%d: \n\t%s\n\t%s\n",
            pNdbConnection->getNdbError(), 
            pNdbConnection->getNdbErrorString(), 
            "NdbConnection");
        pNdb->closeTransaction(pNdbConnection);
        delete pNdb;
    }
    catch(NdbOperation* pNdbOperation)
    {
        printf("%d: \n\t%s\n\t%s\n",
            pNdbOperation->getNdbError(), 
            pNdbOperation->getNdbErrorString(), 
            "NdbOperation");
        pNdb->closeTransaction(pNdbConnection);
        delete pNdb;
    }

    return 0;
}


void Initialize(Ndb* pNdb, long nInsert, bool bStoredTable)
{
    NdbSchemaCon* pNdbSchemaCon;
    NdbSchemaOp* pNdbSchemaOp;
    NdbConnection* pNdbConnection;
    NdbOperation* pNdbOperation;

    try
    {
        _tprintf(_T("Create CallContext table\n"));

        pNdbSchemaCon = pNdb->startSchemaTransaction();
        if(!pNdbSchemaCon)
        {
            throw pNdb;
        }
        pNdbSchemaOp = pNdbSchemaCon->getNdbSchemaOp();
        if(!pNdbSchemaOp)
        {
            throw pNdbSchemaCon;
        }
        if(pNdbSchemaOp->createTable(_T("CallContext"), 8, TupleKey, 2, All, 6, 78, 80, 1, bStoredTable)
            || pNdbSchemaOp->createAttribute(_T("ContextId"), TupleKey, 32, 1, Signed)
            || pNdbSchemaOp->createAttribute(_T("Version"), NoKey, 32, 1, Signed)
            || pNdbSchemaOp->createAttribute(_T("LockFlag"), NoKey, 32, 1, Signed)
            || pNdbSchemaOp->createAttribute(_T("LockTime"), NoKey, 32, 1, Signed)
            || pNdbSchemaOp->createAttribute(_T("LockTimeUSec"), NoKey, 32, 1, Signed)
            || pNdbSchemaOp->createAttribute(_T("ContextData"), NoKey, 8, 4004, String))
        {
            throw pNdbSchemaOp;
        }
        if(pNdbSchemaCon->execute())
        {
            throw pNdbSchemaCon;
        }
        pNdb->closeSchemaTransaction(pNdbSchemaCon);
        
        _tprintf(_T("Insert %d tuples in the CallContext table\n"), nInsert);
        for(long i=0; i<nInsert; ++i)
        {
            long iContextId = -i;
            pNdbConnection = pNdb->startTransaction((Uint32)0, (const char*)&iContextId, (Uint32)4);
            if(!pNdbConnection)
            {
                throw pNdb;
            }
            pNdbOperation = pNdbConnection->getNdbOperation(_T("CallContext"));
            if(!pNdbOperation)
            {
                throw pNdbConnection;
            }
            if(pNdbOperation->insertTuple()
                || pNdbOperation->equal(_T("ContextId"), iContextId)
                || pNdbOperation->setValue(_T("Version"), Int32(1))
                || pNdbOperation->setValue(_T("LockFlag"), Int32(1))
                || pNdbOperation->setValue(_T("LockTime"), Int32(1))
                || pNdbOperation->setValue(_T("LockTimeUSec"), Int32(1))
                || pNdbOperation->setValue(_T("ContextData"), STATUS_DATA, sizeof(STATUS_DATA)))
            {
                throw pNdbOperation;
            }
            if(pNdbConnection->execute(Commit))
            {
                throw pNdbConnection;
            }
            pNdb->closeTransaction(pNdbConnection);
        }
        _tprintf(_T("initialisation done\n"));
    }
    catch(Ndb* pNdb)
    {
        printf("%d: \n\t%s\n\t%s\n",
            pNdb->getNdbError(), 
            pNdb->getNdbErrorString(), 
            "Ndb");
        delete pNdb;
    }
    catch(NdbConnection* pNdbConnection)
    {
        printf("%d: \n\t%s\n\t%s\n",
            pNdbConnection->getNdbError(), 
            pNdbConnection->getNdbErrorString(), 
            "NdbConnection");
        pNdb->closeTransaction(pNdbConnection);
        delete pNdb;
    }
    catch(NdbOperation* pNdbOperation)
    {
        printf("%d: \n\t%s\n\t%s\n",
            pNdbOperation->getNdbError(), 
            pNdbOperation->getNdbErrorString(), 
            "NdbOperation");
        pNdb->closeTransaction(pNdbConnection);
        delete pNdb;
    }
    catch(NdbSchemaCon* pNdbSchemaCon)
    {
        printf("%d: \n\t%s\n\t%s\n",
            pNdbSchemaCon->getNdbError(), 
            pNdbSchemaCon->getNdbErrorString(), 
            "pNdbSchemaCon");
        pNdb->closeSchemaTransaction(pNdbSchemaCon);
        delete pNdb;
    }
    catch(NdbSchemaOp* pNdbSchemaOp)
    {
        printf("%d: \n\t%s\n\t%s\n",
            pNdbSchemaOp->getNdbError(), 
            pNdbSchemaOp->getNdbErrorString(), 
            "pNdbSchemaOp");
        pNdb->closeTransaction(pNdbConnection);
        delete pNdb;
    }
}


int _tmain(int argc, _TCHAR* argv[])
{
	long nNumThreads=4;
	long nSeed = 0;
    long nInsert = 0;
    bool bStoredTable = true;
	if(lstrcmp(argv[1],_T("/?")) == 0)
	{
		_tprintf(_T("InsertRecs [No.Of Threads] [Record Seed No.] [Init no. of rec.] [Stored?]\n"));
		return 0;
	}

	if(argc > 1) 
    	nNumThreads = _ttol(argv[1]);
	else
		nNumThreads = 4;
	if (argc > 2)
		nSeed = _ttol(argv[2]);
	_tprintf(_T("Num of Threads = %d, Seed = %d"), nNumThreads, nSeed);

    if(argc>3)
        nInsert = _ttol(argv[3]);
    if(argc>4)
        bStoredTable = (_ttol(argv[4])!=0);

	long nNumCallsProcessed = 0;

    SetConsoleCtrlHandler(ConsoleCtrlHandler,true); 
	hShutdownEvent = CreateEvent(NULL,TRUE,FALSE,NULL);
   
    // initiate windows sockets
    WORD wVersionRequested;
    WSADATA wsaData;
    int err;
    wVersionRequested = MAKEWORD( 2, 2 );
    err = WSAStartup( wVersionRequested, &wsaData );
    if ( err != 0 ) {
        _tprintf(_T("could not find a usable WinSock DLL\n"));
        return 0;
    }
    if ( LOBYTE( wsaData.wVersion ) != 2 
        || HIBYTE( wsaData.wVersion ) != 2 ) 
    {
        _tprintf(_T("could not find a usable WinSock DLL\n"));
        WSACleanup();
        return 0;
    }

    Ndb* pNdb = new Ndb("TEST_DB");
    if(!pNdb)
    {
        _tprintf(_T("could not construct ndb\n"));
        return 0;
    }
    if(pNdb->init(1)
        || pNdb->waitUntilReady())
    {
        _tprintf(_T("could not initialize ndb\n"));
        return 0;
    }
    
    if(nInsert>0)
    {
        Initialize(pNdb, nInsert, bStoredTable);
    }
    
    if(nNumThreads>0)
    {
        _tprintf(_T("creating %d threads\n"), nNumThreads);
        DWORD dwStartTime  = GetTickCount();
        
        DWORD dwThreadID = 0;
        HANDLE hThreads[50];
        
        struct _ParamStruct params[50];
        
        for(int ij=0;ij<nNumThreads;ij++) {
            params[ij].hShutdownEvent = hShutdownEvent;
            params[ij].nStartingRecordNum = (ij*5000) + nSeed;
            params[ij].pnNumCallsProcessed = &nNumCallsProcessed;
        }
        
        for(ij=0;ij<nNumThreads;ij++) {
            hThreads[ij] = CreateThread(NULL,NULL,RuntimeCallContext,&params[ij],0,&dwThreadID);
        }
        
        //Wait for the threads to finish
        WaitForMultipleObjects(nNumThreads,hThreads,TRUE,INFINITE);	
        DWORD dwEndTime  = GetTickCount();
        
        //Print time taken
        _tprintf(_T("Time Taken for %d Calls is %ld msec (= %ld calls/sec\n"),
            nNumCallsProcessed,dwEndTime-dwStartTime, (1000*nNumCallsProcessed/(dwEndTime-dwStartTime)));
    }

    delete pNdb;
    WSACleanup();
	CloseHandle(hShutdownEvent);
	
    return 0;
}