开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > 数据库系统 > MySQL数据库 > 查看源码
ndb_async2.cpp
资源名称：mysql-4.1.16-win-src.zip　[点击查看]
上传用户：romrleung
上传日期：2022-05-23
资源大小：18897k
文件大小：22k
源码类别：

MySQL数据库

开发平台：

Visual C++

ndb_async2.cpp：源码内容
  1. /* Copyright (C) 2003 MySQL AB
  3.    This program is free software; you can redistribute it and/or modify
  4.    it under the terms of the GNU General Public License as published by
  5.    the Free Software Foundation; either version 2 of the License, or
  6.    (at your option) any later version.
  8.    This program is distributed in the hope that it will be useful,
  9.    but WITHOUT ANY WARRANTY; without even the implied warranty of
  10.    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11.    GNU General Public License for more details.
  13.    You should have received a copy of the GNU General Public License
  14.    along with this program; if not, write to the Free Software
  15.    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
  17. //#define DEBUG_ON
  19. #include <string.h>
  20. #include "userInterface.h"
  22. #include "macros.h"
  23. #include "ndb_schema.hpp"
  24. #include "ndb_error.hpp"
  25. #include <NdbSleep.h>
  27. #include <NdbApi.hpp>
  29. void T1_Callback(int result, NdbConnection * pCon, void * threadData);
  30. void T2_Callback(int result, NdbConnection * pCon, void * threadData);
  31. void T3_Callback_1(int result, NdbConnection * pCon, void * threadData);
  32. void T3_Callback_2(int result, NdbConnection * pCon, void * threadData);
  33. void T3_Callback_3(int result, NdbConnection * pCon, void * threadData);
  34. void T4_Callback_1(int result, NdbConnection * pCon, void * threadData);
  35. void T4_Callback_2(int result, NdbConnection * pCon, void * threadData);
  36. void T4_Callback_3(int result, NdbConnection * pCon, void * threadData);
  37. void T5_Callback_1(int result, NdbConnection * pCon, void * threadData);
  38. void T5_Callback_2(int result, NdbConnection * pCon, void * threadData);
  39. void T5_Callback_3(int result, NdbConnection * pCon, void * threadData);
  41. static int stat_async = 0;
  43. /**
  44.  * Transaction 1 - T1 
  45.  *
  46.  * Update location and changed by/time on a subscriber
  47.  *
  48.  * Input: 
  49.  *   SubscriberNumber,
  50.  *   Location,
  51.  *   ChangedBy,
  52.  *   ChangedTime
  53.  *
  54.  * Output:
  55.  */
  57. #define SFX_START (SUBSCRIBER_NUMBER_LENGTH - SUBSCRIBER_NUMBER_SUFFIX_LENGTH)
  59. inline
  60. NdbConnection *
  61. startTransaction(Ndb * pNDB, ThreadData * td){
  62.   return pNDB->startTransactionDGroup (0, 
  63.        &td->transactionData.number[SFX_START],
  64.        1);
  65. }
  67. void
  68. start_T1(Ndb * pNDB, ThreadData * td, int async){
  70.   DEBUG2("T1(%.*s): - Starting", SUBSCRIBER_NUMBER_LENGTH, 
  71.  td->transactionData.number); 
  73.   NdbConnection * pCON = 0;
  74.   while((pCON = startTransaction(pNDB, td)) == 0){
  75.     CHECK_ALLOWED_ERROR("T1: startTransaction", td, pNDB->getNdbError());
  76.     NdbSleep_MilliSleep(10);
  77.   }
  79.   NdbOperation *MyOp = pCON->getNdbOperation(SUBSCRIBER_TABLE);
  80.   if (MyOp != NULL) {  
  81.     MyOp->updateTuple();  
  82.     MyOp->equal(IND_SUBSCRIBER_NUMBER, 
  83. td->transactionData.number);
  84.     MyOp->setValue(IND_SUBSCRIBER_LOCATION, 
  85.    (char *)&td->transactionData.location);
  86.     MyOp->setValue(IND_SUBSCRIBER_CHANGED_BY, 
  87.    td->transactionData.changed_by);
  88.     MyOp->setValue(IND_SUBSCRIBER_CHANGED_TIME, 
  89.    td->transactionData.changed_time);
  90.     if (async == 1) {
  91.       pCON->executeAsynchPrepare( Commit , T1_Callback, td);
  92.     } else {
  93.       int result = pCON->execute(Commit);
  94.       T1_Callback(result, pCON, (void*)td);
  95.       return;
  96.     }//if
  97.   } else {
  98.     CHECK_NULL(MyOp, "T1: getNdbOperation", td, pCON->getNdbError());
  99.   }//if
  100. }
  102. void
  103. T1_Callback(int result, NdbConnection * pCON, void * threadData) {
  104.   ThreadData * td = (ThreadData *)threadData;
  105.   
  106.   DEBUG2("T1(%.*s): - Completing", SUBSCRIBER_NUMBER_LENGTH, 
  107.  td->transactionData.number); 
  109.   if (result == -1) {
  110.     CHECK_ALLOWED_ERROR("T1: Commit", td, pCON->getNdbError());
  111.     td->pNDB->closeTransaction(pCON);
  112.     start_T1(td->pNDB, td, stat_async);
  113.     return;
  114.   }//if
  115.   td->pNDB->closeTransaction(pCON);
  116.   complete_T1(td);
  117. }
  119. /**
  120.  * Transaction 2 - T2
  121.  *
  122.  * Read from Subscriber:
  123.  *
  124.  * Input: 
  125.  *   SubscriberNumber
  126.  *
  127.  * Output:
  128.  *   Location
  129.  *   Changed by
  130.  *   Changed Timestamp
  131.  *   Name
  132.  */
  133. void
  134. start_T2(Ndb * pNDB, ThreadData * td, int async){
  136.   DEBUG3("T2(%.*s, %d): - Starting", SUBSCRIBER_NUMBER_LENGTH, 
  137.  td->transactionData.number, 
  138.  td->transactionData.location);
  139.   
  140.   NdbConnection * pCON = 0;
  141.   
  142.   while((pCON = startTransaction(pNDB, td)) == 0){
  143.     CHECK_ALLOWED_ERROR("T2-1: startTransaction", td, pNDB->getNdbError());
  144.     NdbSleep_MilliSleep(10);
  145.   }
  147.   NdbOperation *MyOp= pCON->getNdbOperation(SUBSCRIBER_TABLE);
  148.   CHECK_NULL(MyOp, "T2: getNdbOperation", td,
  149.      pCON->getNdbError());
  150.   
  151.   MyOp->readTuple();
  152.   MyOp->equal(IND_SUBSCRIBER_NUMBER,
  153.       td->transactionData.number);
  154.   MyOp->getValue(IND_SUBSCRIBER_LOCATION, 
  155.  (char *)&td->transactionData.location);
  156.   MyOp->getValue(IND_SUBSCRIBER_CHANGED_BY, 
  157.  td->transactionData.changed_by);
  158.   MyOp->getValue(IND_SUBSCRIBER_CHANGED_TIME, 
  159.  td->transactionData.changed_time);
  160.   MyOp->getValue(IND_SUBSCRIBER_NAME, 
  161.  td->transactionData.name);
  162.   if (async == 1) {
  163.     pCON->executeAsynchPrepare( Commit , T2_Callback, td);
  164.   } else {
  165.     int result = pCON->execute(Commit);
  166.     T2_Callback(result, pCON, (void*)td);
  167.     return;
  168.   }//if
  169. }
  171. void
  172. T2_Callback(int result, NdbConnection * pCON, void * threadData){
  173.   ThreadData * td = (ThreadData *)threadData;
  174.   DEBUG3("T2(%.*s, %d): - Completing", SUBSCRIBER_NUMBER_LENGTH, 
  175.  td->transactionData.number, 
  176.  td->transactionData.location);
  177.   
  178.   if (result == -1) {
  179.     CHECK_ALLOWED_ERROR("T2: Commit", td, pCON->getNdbError());
  180.     td->pNDB->closeTransaction(pCON);
  181.     start_T2(td->pNDB, td, stat_async);
  182.     return;
  183.   }//if
  184.   td->pNDB->closeTransaction(pCON);
  185.   complete_T2(td);
  186. }
  188. /**
  189.  * Transaction 3 - T3
  190.  *
  191.  * Read session details
  192.  *
  193.  * Input:
  194.  *   SubscriberNumber
  195.  *   ServerId
  196.  *   ServerBit
  197.  *
  198.  * Output:
  199.  *   BranchExecuted
  200.  *   SessionDetails
  201.  *   ChangedBy
  202.  *   ChangedTime
  203.  *   Location
  204.  */
  205. void
  206. start_T3(Ndb * pNDB, ThreadData * td, int async){
  208.   DEBUG3("T3(%.*s, %.2d): - Starting", SUBSCRIBER_NUMBER_LENGTH, 
  209.  td->transactionData.number, 
  210.  td->transactionData.server_id);
  211.   
  212.   NdbConnection * pCON = 0;
  214.   while((pCON = startTransaction(pNDB, td)) == 0){
  215.     CHECK_ALLOWED_ERROR("T3-1: startTransaction", td, pNDB->getNdbError());
  216.     NdbSleep_MilliSleep(10);
  217.   }
  218.   
  219.   NdbOperation *MyOp= pCON->getNdbOperation(SUBSCRIBER_TABLE);
  220.   CHECK_NULL(MyOp, "T3-1: getNdbOperation", td,
  221.      pCON->getNdbError());
  222.   
  223.   MyOp->readTuple();
  224.   MyOp->equal(IND_SUBSCRIBER_NUMBER, 
  225.       td->transactionData.number);
  226.   MyOp->getValue(IND_SUBSCRIBER_LOCATION, 
  227.  (char *)&td->transactionData.location);
  228.   MyOp->getValue(IND_SUBSCRIBER_CHANGED_BY, 
  229.  td->transactionData.changed_by);
  230.   MyOp->getValue(IND_SUBSCRIBER_CHANGED_TIME, 
  231.  td->transactionData.changed_time);
  232.   MyOp->getValue(IND_SUBSCRIBER_GROUP, 
  233.  (char *)&td->transactionData.group_id);
  234.   MyOp->getValue(IND_SUBSCRIBER_SESSIONS, 
  235.  (char *)&td->transactionData.sessions);
  236.   stat_async = async;
  237.   if (async == 1) {
  238.     pCON->executeAsynchPrepare( NoCommit , T3_Callback_1, td);
  239.   } else {
  240.     int result = pCON->execute( NoCommit );
  241.     T3_Callback_1(result, pCON, (void*)td);
  242.     return;
  243.   }//if
  244. }
  246. void
  247. T3_Callback_1(int result, NdbConnection * pCON, void * threadData){
  248.   ThreadData * td = (ThreadData *)threadData;
  249.   DEBUG3("T3(%.*s, %.2d): - Callback 1", SUBSCRIBER_NUMBER_LENGTH, 
  250.  td->transactionData.number, 
  251.  td->transactionData.server_id);
  253.   if (result == -1) {
  254.     CHECK_ALLOWED_ERROR("T3-1: execute", td, pCON->getNdbError());
  255.     td->pNDB->closeTransaction(pCON);
  256.     start_T3(td->pNDB, td, stat_async);
  257.     return;
  258.   }//if
  260.   NdbOperation * MyOp = pCON->getNdbOperation(GROUP_TABLE);
  261.   CHECK_NULL(MyOp, "T3-2: getNdbOperation", td,
  262.      pCON->getNdbError());
  263.     
  264.   MyOp->readTuple();
  265.   MyOp->equal(IND_GROUP_ID,
  266.       (char*)&td->transactionData.group_id);
  267.   MyOp->getValue(IND_GROUP_ALLOW_READ, 
  268.  (char *)&td->transactionData.permission);
  269.   if (stat_async == 1) {
  270.     pCON->executeAsynchPrepare( NoCommit , T3_Callback_2, td);
  271.   } else {
  272.     int result = pCON->execute( NoCommit );
  273.     T3_Callback_2(result, pCON, (void*)td);
  274.     return;
  275.   }//if
  276. }
  278. void
  279. T3_Callback_2(int result, NdbConnection * pCON, void * threadData){
  280.   ThreadData * td = (ThreadData *)threadData;
  281.   
  282.   if (result == -1) {
  283.     CHECK_ALLOWED_ERROR("T3-2: execute", td, pCON->getNdbError());
  284.     td->pNDB->closeTransaction(pCON);
  285.     start_T3(td->pNDB, td, stat_async);
  286.     return;
  287.   }//if
  288.   
  289.   Uint32 permission = td->transactionData.permission;
  290.   Uint32 sessions   = td->transactionData.sessions;
  291.   Uint32 server_bit = td->transactionData.server_bit;
  293.   if(((permission & server_bit) == server_bit) &&
  294.      ((sessions   & server_bit) == server_bit)){
  295.     
  296.     memcpy(td->transactionData.suffix,
  297.    &td->transactionData.number[SFX_START],
  298.    SUBSCRIBER_NUMBER_SUFFIX_LENGTH);
  299.     DEBUG5("T3(%.*s, %.2d): - Callback 2 - reading(%.*s)", 
  300.    SUBSCRIBER_NUMBER_LENGTH, 
  301.    td->transactionData.number, 
  302.    td->transactionData.server_id,
  303.    SUBSCRIBER_NUMBER_SUFFIX_LENGTH, 
  304.    td->transactionData.suffix);
  305.     
  306.     /* Operation 3 */
  307.     NdbOperation * MyOp = pCON->getNdbOperation(SESSION_TABLE);
  308.     CHECK_NULL(MyOp, "T3-3: getNdbOperation", td,
  309.        pCON->getNdbError());
  310.     
  311.     MyOp->simpleRead();
  312.     MyOp->equal(IND_SESSION_SUBSCRIBER,
  313. (char*)td->transactionData.number);
  314.     MyOp->equal(IND_SESSION_SERVER,
  315. (char*)&td->transactionData.server_id);
  316.     MyOp->getValue(IND_SESSION_DATA, 
  317.    (char *)td->transactionData.session_details);
  318.     
  319.     /* Operation 4 */
  320.     MyOp = pCON->getNdbOperation(SERVER_TABLE);
  321.     CHECK_NULL(MyOp, "T3-4: getNdbOperation", td,
  322.        pCON->getNdbError());
  323.     
  324.     MyOp->interpretedUpdateTuple();
  325.     MyOp->equal(IND_SERVER_ID,
  326. (char*)&td->transactionData.server_id);
  327.     MyOp->equal(IND_SERVER_SUBSCRIBER_SUFFIX,
  328. (char*)td->transactionData.suffix);
  329.     MyOp->incValue(IND_SERVER_READS, (uint32)1);
  330.     td->transactionData.branchExecuted = 1;
  331.   } else {
  332.     DEBUG3("T3(%.*s, %.2d): - Callback 2 - no read",
  333.    SUBSCRIBER_NUMBER_LENGTH, 
  334.    td->transactionData.number, 
  335.    td->transactionData.server_id);
  336.     td->transactionData.branchExecuted = 0;
  337.   }
  338.   if (stat_async == 1) {
  339.     pCON->executeAsynchPrepare( Commit , T3_Callback_3, td);
  340.   } else {
  341.     int result = pCON->execute( Commit );
  342.     T3_Callback_3(result, pCON, (void*)td);
  343.     return;
  344.   }//if
  345. }
  347. void
  348. T3_Callback_3(int result, NdbConnection * pCON, void * threadData){
  349.   ThreadData * td = (ThreadData *)threadData;  
  350.   DEBUG3("T3(%.*s, %.2d): - Completing", SUBSCRIBER_NUMBER_LENGTH, 
  351.  td->transactionData.number, 
  352.  td->transactionData.server_id);
  353.   
  354.   if (result == -1) {
  355.     CHECK_ALLOWED_ERROR("T3-3: Commit", td, pCON->getNdbError());
  356.     td->pNDB->closeTransaction(pCON);
  357.     start_T3(td->pNDB, td, stat_async);
  358.     return;
  359.   }//if
  360.   td->pNDB->closeTransaction(pCON);
  361.   complete_T3(td);
  362. }
  364. /**
  365.  * Transaction 4 - T4
  366.  * 
  367.  * Create session
  368.  *
  369.  * Input:
  370.  *   SubscriberNumber
  371.  *   ServerId
  372.  *   ServerBit
  373.  *   SessionDetails,
  374.  *   DoRollback
  375.  * Output:
  376.  *   ChangedBy
  377.  *   ChangedTime
  378.  *   Location
  379.  *   BranchExecuted
  380.  */
  381. void
  382. start_T4(Ndb * pNDB, ThreadData * td, int async){
  384.   DEBUG3("T4(%.*s, %.2d): - Starting", SUBSCRIBER_NUMBER_LENGTH, 
  385.  td->transactionData.number, 
  386.  td->transactionData.server_id);
  387.   
  388.   NdbConnection * pCON = 0;
  389.   while((pCON = startTransaction(pNDB, td)) == 0){
  390.     CHECK_ALLOWED_ERROR("T4-1: startTransaction", td, pNDB->getNdbError());
  391.     NdbSleep_MilliSleep(10);
  392.   }
  393.   
  394.   NdbOperation *MyOp= pCON->getNdbOperation(SUBSCRIBER_TABLE);
  395.   CHECK_NULL(MyOp, "T4-1: getNdbOperation", td,
  396.      pCON->getNdbError());
  397.   
  398.   MyOp->interpretedUpdateTuple();
  399.   MyOp->equal(IND_SUBSCRIBER_NUMBER, 
  400.       td->transactionData.number);
  401.   MyOp->getValue(IND_SUBSCRIBER_LOCATION, 
  402.  (char *)&td->transactionData.location);
  403.   MyOp->getValue(IND_SUBSCRIBER_CHANGED_BY, 
  404.  td->transactionData.changed_by);
  405.   MyOp->getValue(IND_SUBSCRIBER_CHANGED_TIME, 
  406.  td->transactionData.changed_time);
  407.   MyOp->getValue(IND_SUBSCRIBER_GROUP,
  408.  (char *)&td->transactionData.group_id);
  409.   MyOp->getValue(IND_SUBSCRIBER_SESSIONS,
  410.  (char *)&td->transactionData.sessions); 
  411.   MyOp->incValue(IND_SUBSCRIBER_SESSIONS, 
  412.  (uint32)td->transactionData.server_bit);
  413.   stat_async = async;
  414.   if (async == 1) {
  415.     pCON->executeAsynchPrepare( NoCommit , T4_Callback_1, td);
  416.   } else {
  417.     int result = pCON->execute( NoCommit );
  418.     T4_Callback_1(result, pCON, (void*)td);
  419.     return;
  420.   }//if
  421. }
  423. void
  424. T4_Callback_1(int result, NdbConnection * pCON, void * threadData){
  425.   ThreadData * td = (ThreadData *)threadData;  
  426.   if (result == -1) {
  427.     CHECK_ALLOWED_ERROR("T4-1: execute", td, pCON->getNdbError());
  428.     td->pNDB->closeTransaction(pCON);
  429.     start_T4(td->pNDB, td, stat_async);
  430.     return;
  431.   }//if
  432.   
  433.   DEBUG3("T4(%.*s, %.2d): - Callback 1", 
  434.  SUBSCRIBER_NUMBER_LENGTH, 
  435.  td->transactionData.number, 
  436.  td->transactionData.server_id);
  439.   NdbOperation * MyOp = pCON->getNdbOperation(GROUP_TABLE);
  440.   CHECK_NULL(MyOp, "T4-2: getNdbOperation", td,
  441.      pCON->getNdbError());
  442.   
  443.   MyOp->readTuple();
  444.   MyOp->equal(IND_GROUP_ID,
  445.       (char*)&td->transactionData.group_id);
  446.   MyOp->getValue(IND_GROUP_ALLOW_INSERT, 
  447.  (char *)&td->transactionData.permission);
  448.   if (stat_async == 1) {
  449.     pCON->executeAsynchPrepare( NoCommit , T4_Callback_2, td);
  450.   } else {
  451.     int result = pCON->execute( NoCommit );
  452.     T4_Callback_2(result, pCON, (void*)td);
  453.     return;
  454.   }//if
  455. }
  457. void
  458. T4_Callback_2(int result, NdbConnection * pCON, void * threadData){
  459.   ThreadData * td = (ThreadData *)threadData;  
  460.   if (result == -1) {
  461.     CHECK_ALLOWED_ERROR("T4-2: execute", td, pCON->getNdbError());
  462.     td->pNDB->closeTransaction(pCON);
  463.     start_T4(td->pNDB, td, stat_async);
  464.     return;
  465.   }//if
  467.   Uint32 permission = td->transactionData.permission;
  468.   Uint32 sessions   = td->transactionData.sessions;
  469.   Uint32 server_bit = td->transactionData.server_bit;
  470.   
  471.   if(((permission & server_bit) == server_bit) &&
  472.      ((sessions   & server_bit) == 0)){
  473.     
  474.     memcpy(td->transactionData.suffix,
  475.    &td->transactionData.number[SFX_START],
  476.    SUBSCRIBER_NUMBER_SUFFIX_LENGTH);
  477.     
  478.     DEBUG5("T4(%.*s, %.2d): - Callback 2 - inserting(%.*s)", 
  479.    SUBSCRIBER_NUMBER_LENGTH, 
  480.    td->transactionData.number, 
  481.    td->transactionData.server_id,
  482.    SUBSCRIBER_NUMBER_SUFFIX_LENGTH, 
  483.    td->transactionData.suffix);
  484.     
  485.     /* Operation 3 */
  486.     
  487.     NdbOperation * MyOp = pCON->getNdbOperation(SESSION_TABLE);
  488.     CHECK_NULL(MyOp, "T4-3: getNdbOperation", td,
  489.        pCON->getNdbError());
  490.     
  491.     MyOp->insertTuple();
  492.     MyOp->equal(IND_SESSION_SUBSCRIBER,
  493. (char*)td->transactionData.number);
  494.     MyOp->equal(IND_SESSION_SERVER,
  495. (char*)&td->transactionData.server_id);
  496.     MyOp->setValue(SESSION_DATA, 
  497.    (char *)td->transactionData.session_details);
  498.     /* Operation 4 */
  499.     
  500.     /* Operation 5 */
  501.     MyOp = pCON->getNdbOperation(SERVER_TABLE);
  502.     CHECK_NULL(MyOp, "T4-5: getNdbOperation", td,
  503.        pCON->getNdbError());
  504.     
  505.     MyOp->interpretedUpdateTuple();
  506.     MyOp->equal(IND_SERVER_ID,
  507. (char*)&td->transactionData.server_id);
  508.     MyOp->equal(IND_SERVER_SUBSCRIBER_SUFFIX,
  509. (char*)td->transactionData.suffix);
  510.     MyOp->incValue(IND_SERVER_INSERTS, (uint32)1);
  511.     td->transactionData.branchExecuted = 1;
  512.   } else {
  513.     td->transactionData.branchExecuted = 0;
  514.     DEBUG5("T4(%.*s, %.2d): - Callback 2 - %s %s",
  515.    SUBSCRIBER_NUMBER_LENGTH, 
  516.    td->transactionData.number, 
  517.    td->transactionData.server_id,
  518.    ((permission & server_bit) ? 
  519.     "permission - " : "no permission - "),
  520.    ((sessions   & server_bit) ? 
  521.     "in session - " : "no in session - "));
  522.   }
  523.   
  524.   if(!td->transactionData.do_rollback && td->transactionData.branchExecuted){
  525.     if (stat_async == 1) {
  526.       pCON->executeAsynchPrepare( Commit , T4_Callback_3, td);
  527.     } else {
  528.       int result = pCON->execute( Commit );
  529.       T4_Callback_3(result, pCON, (void*)td);
  530.       return;
  531.     }//if
  532.   } else {
  533.     if (stat_async == 1) {
  534.       pCON->executeAsynchPrepare( Rollback , T4_Callback_3, td);
  535.     } else {
  536.       int result = pCON->execute( Rollback );
  537.       T4_Callback_3(result, pCON, (void*)td);
  538.       return;
  539.     }//if
  540.   }
  541. }
  543. void
  544. T4_Callback_3(int result, NdbConnection * pCON, void * threadData){
  545.   ThreadData * td = (ThreadData *)threadData;  
  546.   if (result == -1) {
  547.     CHECK_ALLOWED_ERROR("T4-3: Commit", td, pCON->getNdbError());
  548.     td->pNDB->closeTransaction(pCON);
  549.     start_T4(td->pNDB, td, stat_async);
  550.     return;
  551.   }//if
  552.   
  553.   DEBUG3("T4(%.*s, %.2d): - Completing", 
  554.  SUBSCRIBER_NUMBER_LENGTH, 
  555.  td->transactionData.number, 
  556.  td->transactionData.server_id);
  558.   td->pNDB->closeTransaction(pCON);
  559.   complete_T4(td);
  560. }
  562. /**
  563.  * Transaction 5 - T5
  564.  * 
  565.  * Delete session
  566.  *
  567.  * Input:
  568.  *   SubscriberNumber
  569.  *   ServerId
  570.  *   ServerBit
  571.  *   DoRollback
  572.  * Output:
  573.  *   ChangedBy
  574.  *   ChangedTime
  575.  *   Location
  576.  *   BranchExecuted
  577.  */
  578. void
  579. start_T5(Ndb * pNDB, ThreadData * td, int async){
  581.   DEBUG3("T5(%.*s, %.2d): - Starting", SUBSCRIBER_NUMBER_LENGTH, 
  582.  td->transactionData.number, 
  583.  td->transactionData.server_id);
  585.   NdbConnection * pCON = 0;
  586.   while((pCON = startTransaction(pNDB, td)) == 0){
  587.     CHECK_ALLOWED_ERROR("T5-1: startTransaction", td, pNDB->getNdbError());
  588.     NdbSleep_MilliSleep(10);
  589.   }
  590.   
  591.   NdbOperation * MyOp= pCON->getNdbOperation(SUBSCRIBER_TABLE);
  592.   CHECK_NULL(MyOp, "T5-1: getNdbOperation", td,
  593.      pCON->getNdbError());
  594.   
  595.   MyOp->interpretedUpdateTuple();
  596.   MyOp->equal(IND_SUBSCRIBER_NUMBER, 
  597.       td->transactionData.number);
  598.   MyOp->getValue(IND_SUBSCRIBER_LOCATION, 
  599.  (char *)&td->transactionData.location);
  600.   MyOp->getValue(IND_SUBSCRIBER_CHANGED_BY, 
  601.  td->transactionData.changed_by);
  602.   MyOp->getValue(IND_SUBSCRIBER_CHANGED_TIME, 
  603.  td->transactionData.changed_time);
  604.   MyOp->getValue(IND_SUBSCRIBER_GROUP,
  605.  (char *)&td->transactionData.group_id);
  606.   MyOp->getValue(IND_SUBSCRIBER_SESSIONS,
  607.  (char *)&td->transactionData.sessions);
  608.   MyOp->subValue(IND_SUBSCRIBER_SESSIONS, 
  609.  (uint32)td->transactionData.server_bit);
  610.   stat_async = async;
  611.   if (async == 1) {
  612.     pCON->executeAsynchPrepare( NoCommit , T5_Callback_1, td);
  613.   } else {
  614.     int result = pCON->execute( NoCommit );
  615.     T5_Callback_1(result, pCON, (void*)td);
  616.     return;
  617.   }//if
  618. }
  620. void
  621. T5_Callback_1(int result, NdbConnection * pCON, void * threadData){
  622.   ThreadData * td = (ThreadData *)threadData;  
  623.   if (result == -1) {
  624.     CHECK_ALLOWED_ERROR("T5-1: execute", td, pCON->getNdbError());
  625.     td->pNDB->closeTransaction(pCON);
  626.     start_T5(td->pNDB, td, stat_async);
  627.     return;
  628.   }//if
  630.   DEBUG3("T5(%.*s, %.2d): - Callback 1", 
  631.  SUBSCRIBER_NUMBER_LENGTH, 
  632.  td->transactionData.number, 
  633.  td->transactionData.server_id);
  634.   
  635.   NdbOperation * MyOp = pCON->getNdbOperation(GROUP_TABLE);
  636.   CHECK_NULL(MyOp, "T5-2: getNdbOperation", td,
  637.      pCON->getNdbError());
  638.   
  639.   MyOp->readTuple();
  640.   MyOp->equal(IND_GROUP_ID,
  641.       (char*)&td->transactionData.group_id);
  642.   MyOp->getValue(IND_GROUP_ALLOW_DELETE, 
  643.  (char *)&td->transactionData.permission);
  644.   if (stat_async == 1) {
  645.     pCON->executeAsynchPrepare( NoCommit , T5_Callback_2, td);
  646.   } else {
  647.     int result = pCON->execute( NoCommit );
  648.     T5_Callback_2(result, pCON, (void*)td);
  649.     return;
  650.   }//if
  651. }
  653. void
  654. T5_Callback_2(int result, NdbConnection * pCON, void * threadData){
  655.   ThreadData * td = (ThreadData *)threadData;  
  656.   if (result == -1) {
  657.     CHECK_ALLOWED_ERROR("T5-2: execute", td, pCON->getNdbError());
  658.     td->pNDB->closeTransaction(pCON);
  659.     start_T5(td->pNDB, td, stat_async);
  660.     return;
  661.   }//if
  663.   Uint32 permission = td->transactionData.permission;
  664.   Uint32 sessions   = td->transactionData.sessions;
  665.   Uint32 server_bit = td->transactionData.server_bit;
  667.   if(((permission & server_bit) == server_bit) &&
  668.      ((sessions   & server_bit) == server_bit)){
  669.     
  670.     memcpy(td->transactionData.suffix,
  671.    &td->transactionData.number[SFX_START],
  672.    SUBSCRIBER_NUMBER_SUFFIX_LENGTH);
  673.     
  674.     DEBUG5("T5(%.*s, %.2d): - Callback 2 - deleting(%.*s)", 
  675.    SUBSCRIBER_NUMBER_LENGTH, 
  676.    td->transactionData.number, 
  677.    td->transactionData.server_id,
  678.    SUBSCRIBER_NUMBER_SUFFIX_LENGTH, 
  679.    td->transactionData.suffix);
  680.     
  681.     /* Operation 3 */
  682.     NdbOperation * MyOp = pCON->getNdbOperation(SESSION_TABLE);
  683.     CHECK_NULL(MyOp, "T5-3: getNdbOperation", td,
  684.        pCON->getNdbError());
  685.     
  686.     MyOp->deleteTuple();
  687.     MyOp->equal(IND_SESSION_SUBSCRIBER,
  688. (char*)td->transactionData.number);
  689.     MyOp->equal(IND_SESSION_SERVER,
  690. (char*)&td->transactionData.server_id);
  691.     /* Operation 4 */
  692.     
  693.     /* Operation 5 */
  694.     MyOp = pCON->getNdbOperation(SERVER_TABLE);
  695.     CHECK_NULL(MyOp, "T5-5: getNdbOperation", td,
  696.        pCON->getNdbError());
  697.     
  698.     MyOp->interpretedUpdateTuple();
  699.     MyOp->equal(IND_SERVER_ID,
  700. (char*)&td->transactionData.server_id);
  701.     MyOp->equal(IND_SERVER_SUBSCRIBER_SUFFIX,
  702. (char*)td->transactionData.suffix);
  703.     MyOp->incValue(IND_SERVER_DELETES, (uint32)1);
  704.     td->transactionData.branchExecuted = 1;
  705.   } else {
  706.     td->transactionData.branchExecuted = 0;
  708.     DEBUG5("T5(%.*s, %.2d): - Callback 2 - no delete - %s %s", 
  709.    SUBSCRIBER_NUMBER_LENGTH, 
  710.    td->transactionData.number, 
  711.    td->transactionData.server_id,
  712.    ((permission & server_bit) ? 
  713.     "permission - " : "no permission - "),
  714.    ((sessions   & server_bit) ? 
  715.     "in session - " : "no in session - "));
  716.   }
  717.   
  718.   if(!td->transactionData.do_rollback && td->transactionData.branchExecuted){
  719.     if (stat_async == 1) {
  720.       pCON->executeAsynchPrepare( Commit , T5_Callback_3, td);
  721.     } else {
  722.       int result = pCON->execute( Commit );
  723.       T5_Callback_3(result, pCON, (void*)td);
  724.       return;
  725.     }//if
  726.   } else {
  727.     if (stat_async == 1) {
  728.       pCON->executeAsynchPrepare( Rollback , T5_Callback_3, td);
  729.     } else {
  730.       int result = pCON->execute( Rollback );
  731.       T5_Callback_3(result, pCON, (void*)td);
  732.       return;
  733.     }//if
  734.   }
  735. }
  737. void
  738. T5_Callback_3(int result, NdbConnection * pCON, void * threadData){
  739.   ThreadData * td = (ThreadData *)threadData;  
  740.   if (result == -1) {
  741.     CHECK_ALLOWED_ERROR("T5-3: Commit", td, pCON->getNdbError());
  742.     td->pNDB->closeTransaction(pCON);
  743.     start_T5(td->pNDB, td, stat_async);
  744.     return;
  745.   }//if
  746.   
  747.   DEBUG3("T5(%.*s, %.2d): - Completing", 
  748.  SUBSCRIBER_NUMBER_LENGTH, 
  749.  td->transactionData.number, 
  750.  td->transactionData.server_id);
  751.   
  752.   td->pNDB->closeTransaction(pCON);
  753.   complete_T5(td);
  754. }