mariadb/ndb/test/ndbapi/ndb_async1.cpp
tomas@poseidon.(none) 8327a68438 neww ndb automake
2004-05-26 15:36:55 +00:00

647 lines
19 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 */
//#define DEBUG_ON
#include "userInterface.h"
#include "macros.h"
#include "ndb_schema.hpp"
#include "ndb_error.hpp"
#include <NdbApi.hpp>
inline
NdbConnection *
startTransaction(Ndb * pNDB,
ServerId inServerId,
const SubscriberNumber inNumber){
const int keyDataLenBytes = sizeof(ServerId)+SUBSCRIBER_NUMBER_LENGTH;
const int keyDataLen_64Words = keyDataLenBytes >> 3;
Uint64 keyDataBuf[keyDataLen_64Words+1]; // The "+1" is for rounding...
char * keyDataBuf_charP = (char *)&keyDataBuf[0];
Uint32 * keyDataBuf_wo32P = (Uint32 *)&keyDataBuf[0];
// Server Id comes first
keyDataBuf_wo32P[0] = inServerId;
// Then subscriber number
memcpy(&keyDataBuf_charP[sizeof(ServerId)], inNumber,
SUBSCRIBER_NUMBER_LENGTH);
return pNDB->startTransaction(0, keyDataBuf_charP, keyDataLenBytes);
}
void T1_Callback(int result, NdbConnection * pCon, void * threadData);
void T2_Callback(int result, NdbConnection * pCon, void * threadData);
void T3_Callback_1(int result, NdbConnection * pCon, void * threadData);
void T3_Callback_2(int result, NdbConnection * pCon, void * threadData);
void T3_Callback_3(int result, NdbConnection * pCon, void * threadData);
void T4_Callback_1(int result, NdbConnection * pCon, void * threadData);
void T4_Callback_2(int result, NdbConnection * pCon, void * threadData);
void T4_Callback_3(int result, NdbConnection * pCon, void * threadData);
void T5_Callback_1(int result, NdbConnection * pCon, void * threadData);
void T5_Callback_2(int result, NdbConnection * pCon, void * threadData);
void T5_Callback_3(int result, NdbConnection * pCon, void * threadData);
/**
* Transaction 1 - T1
*
* Update location and changed by/time on a subscriber
*
* Input:
* SubscriberNumber,
* Location,
* ChangedBy,
* ChangedTime
*
* Output:
*/
void
start_T1(Ndb * pNDB, ThreadData * td){
DEBUG2("T1(%.*s): - Starting\n", SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number);
int check;
NdbConnection * pCON = pNDB->startTransaction();
if (pCON != NULL) {
NdbOperation *MyOp = pCON->getNdbOperation(SUBSCRIBER_TABLE);
if (MyOp != NULL) {
MyOp->updateTuple();
MyOp->equal(IND_SUBSCRIBER_NUMBER,
td->transactionData.number);
MyOp->setValue(IND_SUBSCRIBER_LOCATION,
(char *)&td->transactionData.location);
MyOp->setValue(IND_SUBSCRIBER_CHANGED_BY,
td->transactionData.changed_by);
MyOp->setValue(IND_SUBSCRIBER_CHANGED_TIME,
td->transactionData.changed_time);
pCON->executeAsynchPrepare( Commit , T1_Callback, td);
} else {
CHECK_NULL(MyOp, "T1: getNdbOperation", pCON);
}//if
} else {
error_handler("T1-1: startTranscation",
pNDB->getNdbErrorString(),
pNDB->getNdbError());
}//if
}
void
T1_Callback(int result, NdbConnection * pCON, void * threadData){
ThreadData * td = (ThreadData *)threadData;
DEBUG2("T1(%.*s): - Completing\n", SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number);
CHECK_MINUS_ONE(result, "T1: Commit",
pCON);
td->pNDB->closeTransaction(pCON);
complete_T1(td);
}
/**
* Transaction 2 - T2
*
* Read from Subscriber:
*
* Input:
* SubscriberNumber
*
* Output:
* Location
* Changed by
* Changed Timestamp
* Name
*/
void
start_T2(Ndb * pNDB, ThreadData * td){
DEBUG3("T2(%.*s, %p): - Starting\n", SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.location);
int check;
NdbRecAttr * check2;
NdbConnection * pCON = pNDB->startTransaction();
if (pCON == NULL)
error_handler("T2-1: startTransaction",
pNDB->getNdbErrorString(),
pNDB->getNdbError());
NdbOperation *MyOp= pCON->getNdbOperation(SUBSCRIBER_TABLE);
CHECK_NULL(MyOp, "T2: getNdbOperation",
pCON);
MyOp->readTuple();
MyOp->equal(IND_SUBSCRIBER_NUMBER,
td->transactionData.number);
MyOp->getValue(IND_SUBSCRIBER_LOCATION,
(char *)&td->transactionData.location);
MyOp->getValue(IND_SUBSCRIBER_CHANGED_BY,
td->transactionData.changed_by);
MyOp->getValue(IND_SUBSCRIBER_CHANGED_TIME,
td->transactionData.changed_time);
MyOp->getValue(IND_SUBSCRIBER_NAME,
td->transactionData.name);
pCON->executeAsynchPrepare( Commit, T2_Callback, td );
}
void
T2_Callback(int result, NdbConnection * pCON, void * threadData){
ThreadData * td = (ThreadData *)threadData;
DEBUG3("T2(%.*s, %p): - Completing\n", SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.location);
CHECK_MINUS_ONE(result, "T2: Commit", pCON);
td->pNDB->closeTransaction(pCON);
complete_T2(td);
}
/**
* Transaction 3 - T3
*
* Read session details
*
* Input:
* SubscriberNumber
* ServerId
* ServerBit
*
* Output:
* BranchExecuted
* SessionDetails
* ChangedBy
* ChangedTime
* Location
*/
void
start_T3(Ndb * pNDB, ThreadData * td){
DEBUG3("T3(%.*s, %.2d): - Starting\n", SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.server_id);
int check;
NdbRecAttr * check2;
NdbConnection * pCON = startTransaction(pNDB,
td->transactionData.server_id,
td->transactionData.number);
if (pCON == NULL)
error_handler("T3-1: startTranscation",
pNDB->getNdbErrorString(),
pNDB->getNdbError());
NdbOperation *MyOp= pCON->getNdbOperation(SUBSCRIBER_TABLE);
CHECK_NULL(MyOp, "T3-1: getNdbOperation",
pCON);
MyOp->readTuple();
MyOp->equal(IND_SUBSCRIBER_NUMBER,
td->transactionData.number);
MyOp->getValue(IND_SUBSCRIBER_LOCATION,
(char *)&td->transactionData.location);
MyOp->getValue(IND_SUBSCRIBER_CHANGED_BY,
td->transactionData.changed_by);
MyOp->getValue(IND_SUBSCRIBER_CHANGED_TIME,
td->transactionData.changed_time);
MyOp->getValue(IND_SUBSCRIBER_GROUP,
(char *)&td->transactionData.group_id);
MyOp->getValue(IND_SUBSCRIBER_SESSIONS,
(char *)&td->transactionData.sessions);
pCON->executeAsynchPrepare( NoCommit , T3_Callback_1, td);
}
void
T3_Callback_1(int result, NdbConnection * pCON, void * threadData){
ThreadData * td = (ThreadData *)threadData;
DEBUG3("T3(%.*s, %.2d): - Callback 1\n", SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.server_id);
CHECK_MINUS_ONE(result, "T3-1: NoCommit", pCON);
NdbOperation * MyOp = pCON->getNdbOperation(GROUP_TABLE);
CHECK_NULL(MyOp, "T3-2: getNdbOperation",
pCON);
MyOp->readTuple();
MyOp->equal(IND_GROUP_ID,
(char*)&td->transactionData.group_id);
MyOp->getValue(IND_GROUP_ALLOW_READ,
(char *)&td->transactionData.permission);
pCON->executeAsynchPrepare( NoCommit, T3_Callback_2, td );
}
void
T3_Callback_2(int result, NdbConnection * pCON, void * threadData){
ThreadData * td = (ThreadData *)threadData;
CHECK_MINUS_ONE(result, "T3-2: NoCommit", pCON);
Uint32 permission = td->transactionData.permission;
Uint32 sessions = td->transactionData.sessions;
Uint32 server_bit = td->transactionData.server_bit;
if(((permission & server_bit) == server_bit) &&
((sessions & server_bit) == server_bit)){
memcpy(td->transactionData.suffix,
&td->transactionData.number
[SUBSCRIBER_NUMBER_LENGTH-SUBSCRIBER_NUMBER_SUFFIX_LENGTH],
SUBSCRIBER_NUMBER_SUFFIX_LENGTH);
DEBUG5("T3(%.*s, %.2d): - Callback 2 - reading(%.*s)\n",
SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.server_id,
SUBSCRIBER_NUMBER_SUFFIX_LENGTH,
td->transactionData.suffix);
/* Operation 3 */
NdbOperation * MyOp = pCON->getNdbOperation(SESSION_TABLE);
CHECK_NULL(MyOp, "T3-3: getNdbOperation",
pCON);
MyOp->simpleRead();
MyOp->equal(IND_SESSION_SUBSCRIBER,
(char*)td->transactionData.number);
MyOp->equal(IND_SESSION_SERVER,
(char*)&td->transactionData.server_id);
MyOp->getValue(IND_SESSION_DATA,
(char *)td->transactionData.session_details);
/* Operation 4 */
MyOp = pCON->getNdbOperation(SERVER_TABLE);
CHECK_NULL(MyOp, "T3-4: getNdbOperation",
pCON);
MyOp->interpretedUpdateTuple();
MyOp->equal(IND_SERVER_ID,
(char*)&td->transactionData.server_id);
MyOp->equal(IND_SERVER_SUBSCRIBER_SUFFIX,
(char*)td->transactionData.suffix);
MyOp->incValue(IND_SERVER_READS, (uint32)1);
td->transactionData.branchExecuted = 1;
} else {
DEBUG3("T3(%.*s, %.2d): - Callback 2 - no read\n",
SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.server_id);
td->transactionData.branchExecuted = 0;
}
pCON->executeAsynchPrepare( Commit, T3_Callback_3, td );
}
void
T3_Callback_3(int result, NdbConnection * pCON, void * threadData){
ThreadData * td = (ThreadData *)threadData;
DEBUG3("T3(%.*s, %.2d): - Completing\n", SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.server_id);
CHECK_MINUS_ONE(result, "T3-3: Commit", pCON);
td->pNDB->closeTransaction(pCON);
complete_T3(td);
}
/**
* Transaction 4 - T4
*
* Create session
*
* Input:
* SubscriberNumber
* ServerId
* ServerBit
* SessionDetails,
* DoRollback
* Output:
* ChangedBy
* ChangedTime
* Location
* BranchExecuted
*/
void
start_T4(Ndb * pNDB, ThreadData * td){
DEBUG3("T4(%.*s, %.2d): - Starting\n", SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.server_id);
int check;
NdbRecAttr * check2;
NdbConnection * pCON = startTransaction(pNDB,
td->transactionData.server_id,
td->transactionData.number);
if (pCON == NULL)
error_handler("T4-1: startTranscation",
pNDB->getNdbErrorString(),
pNDB->getNdbError());
NdbOperation *MyOp= pCON->getNdbOperation(SUBSCRIBER_TABLE);
CHECK_NULL(MyOp, "T4-1: getNdbOperation",
pCON);
MyOp->interpretedUpdateTuple();
MyOp->equal(IND_SUBSCRIBER_NUMBER,
td->transactionData.number);
MyOp->getValue(IND_SUBSCRIBER_LOCATION,
(char *)&td->transactionData.location);
MyOp->getValue(IND_SUBSCRIBER_CHANGED_BY,
td->transactionData.changed_by);
MyOp->getValue(IND_SUBSCRIBER_CHANGED_TIME,
td->transactionData.changed_time);
MyOp->getValue(IND_SUBSCRIBER_GROUP,
(char *)&td->transactionData.group_id);
MyOp->getValue(IND_SUBSCRIBER_SESSIONS,
(char *)&td->transactionData.sessions);
MyOp->incValue(IND_SUBSCRIBER_SESSIONS,
(uint32)td->transactionData.server_bit);
pCON->executeAsynchPrepare( NoCommit , T4_Callback_1, td);
}
void
T4_Callback_1(int result, NdbConnection * pCON, void * threadData){
CHECK_MINUS_ONE(result, "T4-1: NoCommit", pCON);
ThreadData * td = (ThreadData *)threadData;
DEBUG3("T4(%.*s, %.2d): - Callback 1\n",
SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.server_id);
NdbOperation * MyOp = pCON->getNdbOperation(GROUP_TABLE);
CHECK_NULL(MyOp, "T4-2: getNdbOperation",
pCON);
MyOp->readTuple();
MyOp->equal(IND_GROUP_ID,
(char*)&td->transactionData.group_id);
MyOp->getValue(IND_GROUP_ALLOW_INSERT,
(char *)&td->transactionData.permission);
pCON->executeAsynchPrepare( NoCommit , T4_Callback_2, td);
}
void
T4_Callback_2(int result, NdbConnection * pCON, void * threadData){
CHECK_MINUS_ONE(result, "T4-2: NoCommit", pCON);
ThreadData * td = (ThreadData *)threadData;
Uint32 permission = td->transactionData.permission;
Uint32 sessions = td->transactionData.sessions;
Uint32 server_bit = td->transactionData.server_bit;
if(((permission & server_bit) == server_bit) &&
((sessions & server_bit) == 0)){
memcpy(td->transactionData.suffix,
&td->transactionData.number
[SUBSCRIBER_NUMBER_LENGTH-SUBSCRIBER_NUMBER_SUFFIX_LENGTH],
SUBSCRIBER_NUMBER_SUFFIX_LENGTH);
DEBUG5("T4(%.*s, %.2d): - Callback 2 - inserting(%.*s)\n",
SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.server_id,
SUBSCRIBER_NUMBER_SUFFIX_LENGTH,
td->transactionData.suffix);
/* Operation 3 */
NdbOperation * MyOp = pCON->getNdbOperation(SESSION_TABLE);
CHECK_NULL(MyOp, "T4-3: getNdbOperation",
pCON);
MyOp->insertTuple();
MyOp->equal(IND_SESSION_SUBSCRIBER,
(char*)td->transactionData.number);
MyOp->equal(IND_SESSION_SERVER,
(char*)&td->transactionData.server_id);
MyOp->setValue(SESSION_DATA,
(char *)td->transactionData.session_details);
/* Operation 4 */
/* Operation 5 */
MyOp = pCON->getNdbOperation(SERVER_TABLE);
CHECK_NULL(MyOp, "T4-5: getNdbOperation",
pCON);
MyOp->interpretedUpdateTuple();
MyOp->equal(IND_SERVER_ID,
(char*)&td->transactionData.server_id);
MyOp->equal(IND_SERVER_SUBSCRIBER_SUFFIX,
(char*)td->transactionData.suffix);
MyOp->incValue(IND_SERVER_INSERTS, (uint32)1);
td->transactionData.branchExecuted = 1;
} else {
td->transactionData.branchExecuted = 0;
DEBUG5("T4(%.*s, %.2d): - Callback 2 - %s %s\n",
SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.server_id,
((permission & server_bit) ?
"permission - " : "no permission - "),
((sessions & server_bit) ?
"in session - " : "no in session - "));
}
if(!td->transactionData.do_rollback && td->transactionData.branchExecuted){
pCON->executeAsynchPrepare(Commit, T4_Callback_3, td);
} else {
pCON->executeAsynchPrepare(Rollback, T4_Callback_3, td);
}
}
void
T4_Callback_3(int result, NdbConnection * pCON, void * threadData){
CHECK_MINUS_ONE(result, "T4-3: Commit", pCON);
ThreadData * td = (ThreadData *)threadData;
DEBUG3("T4(%.*s, %.2d): - Completing\n",
SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.server_id);
td->pNDB->closeTransaction(pCON);
complete_T4(td);
}
/**
* Transaction 5 - T5
*
* Delete session
*
* Input:
* SubscriberNumber
* ServerId
* ServerBit
* DoRollback
* Output:
* ChangedBy
* ChangedTime
* Location
* BranchExecuted
*/
void
start_T5(Ndb * pNDB, ThreadData * td){
DEBUG3("T5(%.*s, %.2d): - Starting\n", SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.server_id);
int check;
NdbRecAttr * check2;
NdbConnection * pCON = pNDB->startTransaction();
if (pCON == NULL)
error_handler("T5-1: startTranscation",
pNDB->getNdbErrorString(),
pNDB->getNdbError());
NdbOperation * MyOp= pCON->getNdbOperation(SUBSCRIBER_TABLE);
CHECK_NULL(MyOp, "T5-1: getNdbOperation",
pCON);
MyOp->interpretedUpdateTuple();
MyOp->equal(IND_SUBSCRIBER_NUMBER,
td->transactionData.number);
MyOp->getValue(IND_SUBSCRIBER_LOCATION,
(char *)&td->transactionData.location);
MyOp->getValue(IND_SUBSCRIBER_CHANGED_BY,
td->transactionData.changed_by);
MyOp->getValue(IND_SUBSCRIBER_CHANGED_TIME,
td->transactionData.changed_time);
MyOp->getValue(IND_SUBSCRIBER_GROUP,
(char *)&td->transactionData.group_id);
MyOp->getValue(IND_SUBSCRIBER_SESSIONS,
(char *)&td->transactionData.sessions);
MyOp->subValue(IND_SUBSCRIBER_SESSIONS,
(uint32)td->transactionData.server_bit);
pCON->executeAsynchPrepare( NoCommit, T5_Callback_1, td );
}
void
T5_Callback_1(int result, NdbConnection * pCON, void * threadData){
CHECK_MINUS_ONE(result, "T5-1: NoCommit", pCON);
ThreadData * td = (ThreadData *)threadData;
DEBUG3("T5(%.*s, %.2d): - Callback 1\n",
SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.server_id);
NdbOperation * MyOp = pCON->getNdbOperation(GROUP_TABLE);
CHECK_NULL(MyOp, "T5-2: getNdbOperation",
pCON);
MyOp->readTuple();
MyOp->equal(IND_GROUP_ID,
(char*)&td->transactionData.group_id);
MyOp->getValue(IND_GROUP_ALLOW_DELETE,
(char *)&td->transactionData.permission);
pCON->executeAsynchPrepare( NoCommit, T5_Callback_2, td );
}
void
T5_Callback_2(int result, NdbConnection * pCON, void * threadData){
CHECK_MINUS_ONE(result, "T5-2: NoCommit", pCON);
ThreadData * td = (ThreadData *)threadData;
Uint32 permission = td->transactionData.permission;
Uint32 sessions = td->transactionData.sessions;
Uint32 server_bit = td->transactionData.server_bit;
if(((permission & server_bit) == server_bit) &&
((sessions & server_bit) == server_bit)){
memcpy(td->transactionData.suffix,
&td->transactionData.number
[SUBSCRIBER_NUMBER_LENGTH-SUBSCRIBER_NUMBER_SUFFIX_LENGTH],
SUBSCRIBER_NUMBER_SUFFIX_LENGTH);
DEBUG5("T5(%.*s, %.2d): - Callback 2 - deleting(%.*s)\n",
SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.server_id,
SUBSCRIBER_NUMBER_SUFFIX_LENGTH,
td->transactionData.suffix);
/* Operation 3 */
NdbOperation * MyOp = pCON->getNdbOperation(SESSION_TABLE);
CHECK_NULL(MyOp, "T5-3: getNdbOperation",
pCON);
MyOp->deleteTuple();
MyOp->equal(IND_SESSION_SUBSCRIBER,
(char*)td->transactionData.number);
MyOp->equal(IND_SESSION_SERVER,
(char*)&td->transactionData.server_id);
/* Operation 4 */
/* Operation 5 */
MyOp = pCON->getNdbOperation(SERVER_TABLE);
CHECK_NULL(MyOp, "T5-5: getNdbOperation",
pCON);
MyOp->interpretedUpdateTuple();
MyOp->equal(IND_SERVER_ID,
(char*)&td->transactionData.server_id);
MyOp->equal(IND_SERVER_SUBSCRIBER_SUFFIX,
(char*)td->transactionData.suffix);
MyOp->incValue(IND_SERVER_DELETES, (uint32)1);
td->transactionData.branchExecuted = 1;
} else {
td->transactionData.branchExecuted = 0;
DEBUG5("T5(%.*s, %.2d): - Callback 2 - no delete - %s %s\n",
SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.server_id,
((permission & server_bit) ?
"permission - " : "no permission - "),
((sessions & server_bit) ?
"in session - " : "no in session - "));
}
if(!td->transactionData.do_rollback && td->transactionData.branchExecuted){
pCON->executeAsynchPrepare(Commit, T5_Callback_3, td);
} else {
pCON->executeAsynchPrepare(Rollback, T5_Callback_3, td);
}
}
void
T5_Callback_3(int result, NdbConnection * pCON, void * threadData){
CHECK_MINUS_ONE(result, "T5-3: Commit", pCON);
ThreadData * td = (ThreadData *)threadData;
DEBUG3("T5(%.*s, %.2d): - Completing\n",
SUBSCRIBER_NUMBER_LENGTH,
td->transactionData.number,
td->transactionData.server_id);
td->pNDB->closeTransaction(pCON);
complete_T5(td);
}