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

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

MySQL数据库

开发平台：

Visual C++

Trix.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. #include "Trix.hpp"
  19. #include <string.h>
  20. #include <kernel_types.h>
  21. #include <NdbOut.hpp>
  23. #include <signaldata/ReadNodesConf.hpp>
  24. #include <signaldata/NodeFailRep.hpp>
  25. #include <signaldata/DumpStateOrd.hpp>
  26. #include <signaldata/GetTabInfo.hpp>
  27. #include <signaldata/DictTabInfo.hpp>
  28. #include <signaldata/BuildIndx.hpp>
  29. #include <signaldata/SumaImpl.hpp>
  30. #include <signaldata/UtilPrepare.hpp>
  31. #include <signaldata/UtilExecute.hpp>
  32. #include <signaldata/UtilRelease.hpp>
  33. #include <SectionReader.hpp>
  34. #include <AttributeHeader.hpp>
  36. #define CONSTRAINT_VIOLATION 893
  38. #define DEBUG(x) { ndbout << "TRIX::" << x << endl; }
  40. /**
  41.  *
  42.  */
  43. Trix::Trix(const Configuration & conf) :
  44.   SimulatedBlock(TRIX, conf),
  45.   c_theNodes(c_theNodeRecPool),
  46.   c_masterNodeId(0),
  47.   c_masterTrixRef(0),
  48.   c_noNodesFailed(0),
  49.   c_noActiveNodes(0),
  50.   c_theSubscriptions(c_theSubscriptionRecPool)
  51. {
  52.   BLOCK_CONSTRUCTOR(Trix);
  54.   // Add received signals
  55.   addRecSignal(GSN_STTOR,  &Trix::execSTTOR);
  56.   addRecSignal(GSN_NDB_STTOR,  &Trix::execNDB_STTOR); // Forwarded from DICT
  57.   addRecSignal(GSN_READ_NODESCONF, &Trix::execREAD_NODESCONF);
  58.   addRecSignal(GSN_READ_NODESREF, &Trix::execREAD_NODESREF);
  59.   addRecSignal(GSN_NODE_FAILREP, &Trix::execNODE_FAILREP);
  60.   addRecSignal(GSN_INCL_NODEREQ, &Trix::execINCL_NODEREQ);
  61.   addRecSignal(GSN_DUMP_STATE_ORD, &Trix::execDUMP_STATE_ORD);
  63.   // Index build
  64.   addRecSignal(GSN_BUILDINDXREQ, &Trix::execBUILDINDXREQ);
  65.   // Dump testing
  66.   addRecSignal(GSN_BUILDINDXCONF, &Trix::execBUILDINDXCONF);
  67.   addRecSignal(GSN_BUILDINDXREF, &Trix::execBUILDINDXREF);
  70.   addRecSignal(GSN_UTIL_PREPARE_CONF, &Trix::execUTIL_PREPARE_CONF);
  71.   addRecSignal(GSN_UTIL_PREPARE_REF, &Trix::execUTIL_PREPARE_REF);
  72.   addRecSignal(GSN_UTIL_EXECUTE_CONF, &Trix::execUTIL_EXECUTE_CONF);
  73.   addRecSignal(GSN_UTIL_EXECUTE_REF, &Trix::execUTIL_EXECUTE_REF);
  74.   addRecSignal(GSN_UTIL_RELEASE_CONF, &Trix::execUTIL_RELEASE_CONF);
  75.   addRecSignal(GSN_UTIL_RELEASE_REF, &Trix::execUTIL_RELEASE_REF);
  78.   // Suma signals
  79.   addRecSignal(GSN_SUB_CREATE_CONF, &Trix::execSUB_CREATE_CONF);
  80.   addRecSignal(GSN_SUB_CREATE_REF, &Trix::execSUB_CREATE_REF);
  81.   addRecSignal(GSN_SUB_REMOVE_CONF, &Trix::execSUB_REMOVE_CONF);
  82.   addRecSignal(GSN_SUB_REMOVE_REF, &Trix::execSUB_REMOVE_REF);
  83.   addRecSignal(GSN_SUB_SYNC_CONF, &Trix::execSUB_SYNC_CONF);
  84.   addRecSignal(GSN_SUB_SYNC_REF, &Trix::execSUB_SYNC_REF);
  85.   addRecSignal(GSN_SUB_SYNC_CONTINUE_REQ, &Trix::execSUB_SYNC_CONTINUE_REQ);
  86.   addRecSignal(GSN_SUB_META_DATA, &Trix::execSUB_META_DATA);
  87.   addRecSignal(GSN_SUB_TABLE_DATA, &Trix::execSUB_TABLE_DATA);
  89.   // Allocate pool sizes
  90.   c_theAttrOrderBufferPool.setSize(100);
  91.   c_theSubscriptionRecPool.setSize(100);
  93.   ArrayList<SubscriptionRecord> subscriptions(c_theSubscriptionRecPool);
  94.   SubscriptionRecPtr subptr;
  95.   while(subscriptions.seize(subptr) == true) {
  96.     new (subptr.p) SubscriptionRecord(c_theAttrOrderBufferPool);
  97.   }
  98.   subscriptions.release();
  99. }
  101. /**
  102.  *
  103.  */
  104. Trix::~Trix()
  105. {
  106. }
  108. /**
  109.  *
  110.  */
  111. void Trix::execSTTOR(Signal* signal) 
  112. {
  113.   jamEntry();                            
  115.   //const Uint32 startphase   = signal->theData[1];
  116.   const Uint32 theSignalKey = signal->theData[6];
  117.   
  118.   signal->theData[0] = theSignalKey;
  119.   signal->theData[3] = 1;
  120.   signal->theData[4] = 255; // No more start phases from missra
  121.   sendSignal(NDBCNTR_REF, GSN_STTORRY, signal, 5, JBB);
  122.   return;
  123. }//Trix::execSTTOR()
  125. /**
  126.  *
  127.  */
  128. void Trix::execNDB_STTOR(Signal* signal) 
  129. {
  130.   jamEntry();                            
  131.   BlockReference ndbcntrRef = signal->theData[0];  
  132.   Uint16 startphase = signal->theData[2];      /* RESTART PHASE           */
  133.   Uint16 mynode = signal->theData[1];  
  134.   //Uint16 restarttype = signal->theData[3];  
  135.   //UintR configInfo1 = signal->theData[6];     /* CONFIGRATION INFO PART 1 */
  136.   //UintR configInfo2 = signal->theData[7];     /* CONFIGRATION INFO PART 2 */
  137.   switch (startphase) {
  138.   case 3:
  139.     jam();
  140.     /* SYMBOLIC START PHASE 4             */
  141.     /* ABSOLUTE PHASE 5                   */
  142.     /* REQUEST NODE IDENTITIES FROM DBDIH */
  143.     signal->theData[0] = calcTrixBlockRef(mynode);
  144.     sendSignal(ndbcntrRef, GSN_READ_NODESREQ, signal, 1, JBB);
  145.     return;
  146.     break;
  147.   case 6:
  148.     break;
  149.   default:
  150.     break;
  151.   }
  152. }
  154. /**
  155.  *
  156.  */
  157. void Trix::execREAD_NODESCONF(Signal* signal)
  158. {
  159.   jamEntry();
  161.   ReadNodesConf * const  readNodes = (ReadNodesConf *)signal->getDataPtr();
  162.   //Uint32 noOfNodes   = readNodes->noOfNodes;
  163.   NodeRecPtr nodeRecPtr;
  165.   c_masterNodeId = readNodes->masterNodeId;
  166.   c_masterTrixRef = RNIL;
  167.   c_noNodesFailed = 0;
  169.   for(unsigned i = 0; i < MAX_NDB_NODES; i++) {
  170.     jam();
  171.     if(NodeBitmask::get(readNodes->allNodes, i)) {
  172.       // Node is defined
  173.       jam();
  174.       ndbrequire(c_theNodes.seizeId(nodeRecPtr, i));
  175.       nodeRecPtr.p->trixRef = calcTrixBlockRef(i);
  176.       if (i == c_masterNodeId) {
  177.         c_masterTrixRef = nodeRecPtr.p->trixRef;
  178.       }
  179.       if(NodeBitmask::get(readNodes->inactiveNodes, i)){
  180.         // Node is not active
  181. jam();
  182. /**-----------------------------------------------------------------
  183.  * THIS NODE IS DEFINED IN THE CLUSTER BUT IS NOT ALIVE CURRENTLY.
  184.  * WE ADD THE NODE TO THE SET OF FAILED NODES AND ALSO SET THE
  185.  * BLOCKSTATE TO BUSY TO AVOID ADDING TRIGGERS OR INDEXES WHILE 
  186.  * NOT ALL NODES ARE ALIVE.
  187.  *------------------------------------------------------------------*/
  188. arrGuard(c_noNodesFailed, MAX_NDB_NODES);
  189. nodeRecPtr.p->alive = false;
  190. c_noNodesFailed++;
  191. c_blockState = Trix::NODE_FAILURE;
  192.       }
  193.       else {
  194.         // Node is active
  195.         jam();
  196.         c_noActiveNodes++;
  197.         nodeRecPtr.p->alive = true;
  198.       }
  199.     }
  200.   }
  201.   if (c_noNodesFailed == 0) {
  202.     c_blockState = Trix::STARTED;
  203.   }
  204. }
  206. /**
  207.  *
  208.  */
  209. void Trix::execREAD_NODESREF(Signal* signal)
  210. {
  211.   // NYI
  212. }
  214. /**
  215.  *
  216.  */
  217. void Trix::execNODE_FAILREP(Signal* signal)
  218. {
  219.   jamEntry();
  220.   NodeFailRep * const  nodeFail = (NodeFailRep *) signal->getDataPtr();
  222.   //Uint32 failureNr    = nodeFail->failNo;
  223.   //Uint32 numberNodes  = nodeFail->noOfNodes;
  224.   Uint32 masterNodeId = nodeFail->masterNodeId;
  226.   NodeRecPtr nodeRecPtr;
  228.   for(c_theNodes.first(nodeRecPtr); 
  229.       nodeRecPtr.i != RNIL; 
  230.       c_theNodes.next(nodeRecPtr)) {
  231.     if(NodeBitmask::get(nodeFail->theNodes, nodeRecPtr.i)) {
  232.       nodeRecPtr.p->alive = false;
  233.       c_noNodesFailed++;
  234.       c_noActiveNodes--;      
  235.     }
  236.   }
  237.   if (c_masterNodeId != masterNodeId) {
  238.     c_masterNodeId = masterNodeId;
  239.     NodeRecord* nodeRec = c_theNodes.getPtr(masterNodeId);
  240.     c_masterTrixRef = nodeRec->trixRef;
  241.   }
  242. }
  244. /**
  245.  *
  246.  */
  247. void Trix::execINCL_NODEREQ(Signal* signal)
  248. {
  249.   jamEntry();
  250.   UintR node_id = signal->theData[1];
  251.   NodeRecord* nodeRec = c_theNodes.getPtr(node_id);
  252.   nodeRec->alive = true;
  253.   c_noNodesFailed--;
  254.   c_noActiveNodes++;      
  255.   nodeRec->trixRef = calcTrixBlockRef(node_id);
  256.   if (c_noNodesFailed == 0) {
  257.     c_blockState = Trix::STARTED;
  258.   }  
  259. }
  261. // Debugging
  262. void
  263. Trix::execDUMP_STATE_ORD(Signal* signal)
  264. {
  265.   jamEntry();
  267.   DumpStateOrd * dumpStateOrd = (DumpStateOrd *)signal->getDataPtr();
  269.   switch(dumpStateOrd->args[0]) {
  270.   case(300): {// ok
  271.     // index2 -T; index2 -I -n10000; index2 -c
  272.     // all dump 300 0 0 0 0 0 4 2
  273.     // select_count INDEX0000
  274.     BuildIndxReq * buildIndxReq = (BuildIndxReq *)signal->getDataPtrSend();
  275.     
  276.     MEMCOPY_NO_WORDS(buildIndxReq, 
  277.      signal->theData + 1, 
  278.      BuildIndxReq::SignalLength);
  279.     buildIndxReq->setUserRef(reference()); // return to me
  280.     buildIndxReq->setParallelism(10);
  281.     Uint32 indexColumns[1] = {1};
  282.     Uint32 keyColumns[1] = {0};
  283.     struct LinearSectionPtr orderPtr[2];
  284.     buildIndxReq->setColumnOrder(indexColumns, 1, keyColumns, 1, orderPtr);
  285.     sendSignal(reference(),
  286.        GSN_BUILDINDXREQ,
  287.        signal,
  288.        BuildIndxReq::SignalLength,
  289.        JBB,
  290.        orderPtr,
  291.        BuildIndxReq::NoOfSections);
  292.     break;
  293.   }
  294.   case(301): { // ok
  295.     // index2 -T; index2 -I -n10000; index2 -c -p
  296.     // all dump 301 0 0 0 0 0 4 2
  297.     // select_count INDEX0000
  298.     BuildIndxReq * buildIndxReq = (BuildIndxReq *)signal->getDataPtrSend();
  299.     
  300.     MEMCOPY_NO_WORDS(buildIndxReq, 
  301.      signal->theData + 1, 
  302.      BuildIndxReq::SignalLength);
  303.     buildIndxReq->setUserRef(reference()); // return to me
  304.     buildIndxReq->setParallelism(10);
  305.     Uint32 indexColumns[2] = {0, 1};
  306.     Uint32 keyColumns[1] = {0};
  307.     struct LinearSectionPtr orderPtr[2];
  308.     buildIndxReq->setColumnOrder(indexColumns, 2, keyColumns, 1, orderPtr);
  309.     sendSignal(reference(),
  310.        GSN_BUILDINDXREQ,
  311.        signal,
  312.        BuildIndxReq::SignalLength,
  313.        JBB,
  314.        orderPtr,
  315.        BuildIndxReq::NoOfSections);
  316.     break;
  317.   }
  318.   case(302): { // ok
  319.     // index -T; index -I -n1000; index -c -p
  320.     // all dump 302 0 0 0 0 0 4 2
  321.     // select_count PNUMINDEX0000
  322.     BuildIndxReq * buildIndxReq = (BuildIndxReq *)signal->getDataPtrSend();
  323.     
  324.     MEMCOPY_NO_WORDS(buildIndxReq, 
  325.      signal->theData + 1, 
  326.      BuildIndxReq::SignalLength);
  327.     buildIndxReq->setUserRef(reference()); // return to me
  328.     buildIndxReq->setParallelism(10);
  329.     Uint32 indexColumns[3] = {0, 3, 5};
  330.     Uint32 keyColumns[1] = {0};
  331.     struct LinearSectionPtr orderPtr[2];
  332.     buildIndxReq->setColumnOrder(indexColumns, 3, keyColumns, 1, orderPtr);
  333.     sendSignal(reference(),
  334.        GSN_BUILDINDXREQ,
  335.        signal,
  336.        BuildIndxReq::SignalLength,
  337.        JBB,
  338.        orderPtr,
  339.        BuildIndxReq::NoOfSections);
  340.     break;
  341.   }
  342.   case(303): { // ok
  343.     // index -T -2; index -I -2 -n1000; index -c -p
  344.     // all dump 303 0 0 0 0 0 4 2
  345.     // select_count PNUMINDEX0000
  346.     BuildIndxReq * buildIndxReq = (BuildIndxReq *)signal->getDataPtrSend();
  347.     
  348.     MEMCOPY_NO_WORDS(buildIndxReq, 
  349.      signal->theData + 1, 
  350.      BuildIndxReq::SignalLength);
  351.     buildIndxReq->setUserRef(reference()); // return to me
  352.     buildIndxReq->setParallelism(10);
  353.     Uint32 indexColumns[3] = {0, 3, 5};
  354.     Uint32 keyColumns[2] = {0, 1};
  355.     struct LinearSectionPtr orderPtr[2];
  356.     buildIndxReq->setColumnOrder(indexColumns, 3, keyColumns, 2, orderPtr);
  357.     sendSignal(reference(),
  358.        GSN_BUILDINDXREQ,
  359.        signal,
  360.        BuildIndxReq::SignalLength,
  361.        JBB,
  362.        orderPtr,
  363.        BuildIndxReq::NoOfSections);
  364.     break;
  365.   }
  366.   case(304): { // ok
  367.     // index -T -L; index -I -L -n1000; index -c -p
  368.     // all dump 304 0 0 0 0 0 4 2
  369.     // select_count PNUMINDEX0000
  370.     BuildIndxReq * buildIndxReq = (BuildIndxReq *)signal->getDataPtrSend();
  371.     
  372.     MEMCOPY_NO_WORDS(buildIndxReq, 
  373.      signal->theData + 1, 
  374.      BuildIndxReq::SignalLength);
  375.     buildIndxReq->setUserRef(reference()); // return to me
  376.     buildIndxReq->setParallelism(10);
  377.     Uint32 indexColumns[3] = {0, 3, 5};
  378.     Uint32 keyColumns[1] = {0};
  379.     struct LinearSectionPtr orderPtr[2];
  380.     buildIndxReq->setColumnOrder(indexColumns, 3, keyColumns, 1, orderPtr);
  381.     sendSignal(reference(),
  382.        GSN_BUILDINDXREQ,
  383.        signal,
  384.        BuildIndxReq::SignalLength,
  385.        JBB,
  386.        orderPtr,
  387.        BuildIndxReq::NoOfSections);
  388.     break;
  389.   }
  390.   case(305): { // ok
  391.     // index -T -2 -L; index -I -2 -L -n1000; index -c -p
  392.     // all dump 305 0 0 0 0 0 4 2
  393.     // select_count PNUMINDEX0000
  394.     BuildIndxReq * buildIndxReq = (BuildIndxReq *)signal->getDataPtrSend();
  395.     
  396.     MEMCOPY_NO_WORDS(buildIndxReq, 
  397.      signal->theData + 1, 
  398.      BuildIndxReq::SignalLength);
  399.     buildIndxReq->setUserRef(reference()); // return to me
  400.     buildIndxReq->setParallelism(10);
  401.     Uint32 indexColumns[3] = {0, 3, 5};
  402.     Uint32 keyColumns[2] = {0, 1};
  403.     struct LinearSectionPtr orderPtr[2];
  404.     buildIndxReq->setColumnOrder(indexColumns, 3, keyColumns, 2, orderPtr);
  405.     sendSignal(reference(),
  406.        GSN_BUILDINDXREQ,
  407.        signal,
  408.        BuildIndxReq::SignalLength,
  409.        JBB,
  410.        orderPtr,
  411.        BuildIndxReq::NoOfSections);
  412.     break;
  413.   }
  414.   default: {
  415.     // Ignore
  416.   }
  417.   }
  418. }
  420. // Build index
  421. /**
  422.  *
  423.  */
  424. void Trix:: execBUILDINDXREQ(Signal* signal)
  425. {
  426.   jamEntry();
  427.   BuildIndxReq * buildIndxReq = (BuildIndxReq *)signal->getDataPtr();
  429.   // Seize a subscription record
  430.   SubscriptionRecPtr subRecPtr;
  431.   SubscriptionRecord* subRec;
  432.   
  433.   if (!c_theSubscriptions.seizeId(subRecPtr, buildIndxReq->getBuildId())) {
  434.     // Failed to allocate subscription record
  435.     BuildIndxRef * buildIndxRef = (BuildIndxRef *)signal->getDataPtrSend();
  437.     buildIndxRef->setErrorCode(BuildIndxRef::AllocationFailure);
  438.     releaseSections(signal);
  439.     sendSignal(buildIndxReq->getUserRef(), 
  440.        GSN_BUILDINDXREF, signal, BuildIndxRef::SignalLength, JBB);
  441.     return;
  442.   }
  443.   subRec = subRecPtr.p;
  444.   subRec->errorCode = BuildIndxRef::NoError;
  445.   subRec->userReference = buildIndxReq->getUserRef();
  446.   subRec->connectionPtr = buildIndxReq->getConnectionPtr();
  447.   subRec->subscriptionId = buildIndxReq->getBuildId();
  448.   subRec->subscriptionKey = buildIndxReq->getBuildKey();
  449.   subRec->indexType = buildIndxReq->getIndexType();
  450.   subRec->sourceTableId = buildIndxReq->getTableId();
  451.   subRec->targetTableId = buildIndxReq->getIndexId();
  452.   subRec->parallelism = buildIndxReq->getParallelism();
  453.   subRec->expectedConf = 0;
  454.   subRec->subscriptionCreated = false;
  455.   subRec->pendingSubSyncContinueConf = false;
  456.   subRec->prepareId = RNIL;
  458.   // Get column order segments
  459.   Uint32 noOfSections = signal->getNoOfSections();
  460.   if(noOfSections > 0) {
  461.     SegmentedSectionPtr ptr;
  462.     signal->getSection(ptr, BuildIndxReq::INDEX_COLUMNS);
  463.     append(subRec->attributeOrder, ptr, getSectionSegmentPool());
  464.     subRec->noOfIndexColumns = ptr.sz;
  465.   }
  466.   if(noOfSections > 1) {
  467.     SegmentedSectionPtr ptr;
  468.     signal->getSection(ptr, BuildIndxReq::KEY_COLUMNS);
  469.     append(subRec->attributeOrder, ptr, getSectionSegmentPool());
  470.     subRec->noOfKeyColumns = ptr.sz;
  471.   }
  472. #if 0
  473.   // Debugging
  474.   printf("Trix:: execBUILDINDXREQ: Attribute order:n");
  475.   subRec->attributeOrder.print(stdout);
  476. #endif
  477.   releaseSections(signal);
  478.   prepareInsertTransactions(signal, subRecPtr);
  479. }
  481. void Trix:: execBUILDINDXCONF(Signal* signal)
  482. {
  483.   printf("Trix:: execBUILDINDXCONFn");
  484. }
  486. void Trix:: execBUILDINDXREF(Signal* signal)
  487. {
  488.   printf("Trix:: execBUILDINDXREFn");
  489. }
  491. void Trix::execUTIL_PREPARE_CONF(Signal* signal)
  492. {
  493.   jamEntry();
  494.   UtilPrepareConf * utilPrepareConf = (UtilPrepareConf *)signal->getDataPtr();
  495.   SubscriptionRecPtr subRecPtr;
  496.   SubscriptionRecord* subRec;
  498.   subRecPtr.i = utilPrepareConf->senderData;
  499.   if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
  500.     printf("Trix::execUTIL_PREPARE_CONF: Failed to find subscription data %un", subRecPtr.i);
  501.     return;
  502.   }
  503.   subRecPtr.p = subRec;
  504.   subRec->prepareId = utilPrepareConf->prepareId;
  505.   setupSubscription(signal, subRecPtr);
  506. }
  508. void Trix::execUTIL_PREPARE_REF(Signal* signal)
  509. {
  510.   jamEntry();
  511.   UtilPrepareRef * utilPrepareRef = (UtilPrepareRef *)signal->getDataPtr();
  512.   SubscriptionRecPtr subRecPtr;
  513.   SubscriptionRecord* subRec;
  515.   subRecPtr.i = utilPrepareRef->senderData;
  516.   if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
  517.     printf("Trix::execUTIL_PREPARE_REF: Failed to find subscription data %un", subRecPtr.i);
  518.     return;
  519.   }
  520.   subRecPtr.p = subRec;
  521.   subRec->errorCode = BuildIndxRef::InternalError;
  522. }
  524. void Trix::execUTIL_EXECUTE_CONF(Signal* signal)
  525. {
  526.   jamEntry();
  527.   UtilExecuteConf * utilExecuteConf = (UtilExecuteConf *)signal->getDataPtr();
  528.   SubscriptionRecPtr subRecPtr;
  529.   SubscriptionRecord* subRec;
  531.   subRecPtr.i = utilExecuteConf->senderData;
  532.   if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
  533.     printf("rix::execUTIL_EXECUTE_CONF: Failed to find subscription data %un", subRecPtr.i);
  534.     return;
  535.   }
  536.   subRecPtr.p = subRec;
  537.   subRec->expectedConf--;
  538.   checkParallelism(signal, subRec);
  539.   if (subRec->expectedConf == 0)
  540.     buildComplete(signal, subRecPtr);
  541. }
  543. void Trix::execUTIL_EXECUTE_REF(Signal* signal)
  544. {
  545.   jamEntry();
  546.   UtilExecuteRef * utilExecuteRef = (UtilExecuteRef *)signal->getDataPtr();
  547.   SubscriptionRecPtr subRecPtr;
  548.   SubscriptionRecord* subRec;
  550.   subRecPtr.i = utilExecuteRef->senderData;
  551.   if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
  552.     printf("Trix::execUTIL_EXECUTE_REF: Failed to find subscription data %un", subRecPtr.i);
  553.     return;
  554.   }
  555.   subRecPtr.p = subRec;
  556.   ndbrequire(utilExecuteRef->errorCode == UtilExecuteRef::TCError);
  557.   if(utilExecuteRef->TCErrorCode == CONSTRAINT_VIOLATION)
  558.     buildFailed(signal, subRecPtr, BuildIndxRef::IndexNotUnique);
  559.   else
  560.     buildFailed(signal, subRecPtr, BuildIndxRef::InternalError);
  561. }
  563. void Trix::execSUB_CREATE_CONF(Signal* signal)
  564. {
  565.   jamEntry();
  566.   SubCreateConf * subCreateConf = (SubCreateConf *)signal->getDataPtr();
  567.   SubscriptionRecPtr subRecPtr;
  568.   SubscriptionRecord* subRec;
  570.   subRecPtr.i = subCreateConf->subscriberData;
  571.   if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
  572.     printf("Trix::execSUB_CREATE_CONF: Failed to find subscription data %un", subRecPtr.i);
  573.     return;
  574.   }
  575.   ndbrequire(subRec->subscriptionId == subCreateConf->subscriptionId);
  576.   ndbrequire(subRec->subscriptionKey == subCreateConf->subscriptionKey);
  577.   subRec->subscriptionCreated = true;
  578.   subRecPtr.p = subRec;
  579.   setupTableScan(signal, subRecPtr);
  580. }
  582. void Trix::execSUB_CREATE_REF(Signal* signal)
  583. {
  584.   jamEntry();
  585.   // THIS SIGNAL IS NEVER SENT FROM SUMA?
  586.   /*
  587.   SubCreateRef * subCreateRef = (SubCreateRef *)signal->getDataPtr();
  588.   SubscriptionRecPtr subRecPtr;
  589.   SubscriptionRecord* subRec;
  591.   subRecPtr.i = subCreateRef->subscriberData;
  592.   if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
  593.     printf("Trix::execSUB_CREATE_REF: Failed to find subscription data %un", subRecPtr.i);
  594.     return;
  595.   }
  596.   subRecPtr.p = subRec;
  597.   buildFailed(signal, subRecPtr, BuildIndxRef::InternalError);
  598.   */
  599. }
  601. void Trix::execSUB_SYNC_CONF(Signal* signal)
  602. {
  603.   jamEntry();
  604.   SubSyncConf * subSyncConf = (SubSyncConf *)signal->getDataPtr();
  605.   SubscriptionRecPtr subRecPtr;
  606.   SubscriptionRecord* subRec;
  607.   
  608.   subRecPtr.i = subSyncConf->subscriberData;
  609.   if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
  610.     printf("Trix::execSUB_SYNC_CONF: Failed to find subscription data %un", subRecPtr.i);
  611.     return;
  612.   }
  613.   ndbrequire(subRec->subscriptionId == subSyncConf->subscriptionId);
  614.   ndbrequire(subRec->subscriptionKey == subSyncConf->subscriptionKey);
  615.   subRecPtr.p = subRec;
  616.   if(subSyncConf->part == SubscriptionData::MetaData)
  617.     startTableScan(signal, subRecPtr);
  618.   else {
  619.     subRec->expectedConf--;
  620.     checkParallelism(signal, subRec);
  621.     if (subRec->expectedConf == 0)
  622.       buildComplete(signal, subRecPtr);
  623.   }
  624. }
  626. void Trix::execSUB_SYNC_REF(Signal* signal)
  627. {
  628.   jamEntry();
  629.   SubSyncRef * subSyncRef = (SubSyncRef *)signal->getDataPtr();
  630.   SubscriptionRecPtr subRecPtr;
  631.   SubscriptionRecord* subRec;
  633.   subRecPtr.i = subSyncRef->subscriberData;
  634.   if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
  635.     printf("Trix::execSUB_SYNC_REF: Failed to find subscription data %un", subRecPtr.i);
  636.     return;
  637.   }
  638.   subRecPtr.p = subRec;
  639.   buildFailed(signal, subRecPtr, BuildIndxRef::InternalError);
  640. }
  642. void Trix::execSUB_SYNC_CONTINUE_REQ(Signal* signal)
  643. {
  644.   SubSyncContinueReq  * subSyncContinueReq = 
  645.     (SubSyncContinueReq *) signal->getDataPtr();
  646.   
  647.   SubscriptionRecPtr subRecPtr;
  648.   SubscriptionRecord* subRec;
  649.   subRecPtr.i = subSyncContinueReq->subscriberData;
  650.   if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
  651.     printf("Trix::execSUB_SYNC_CONTINUE_REQ: Failed to find subscription data %un", subRecPtr.i);
  652.     return;
  653.   }
  654.   subRecPtr.p = subRec;
  655.   subRec->pendingSubSyncContinueConf = true;
  656.   checkParallelism(signal, subRec);
  657. }
  659. void Trix::execSUB_META_DATA(Signal* signal)
  660. {
  661.   jamEntry();
  662. }
  664. void Trix::execSUB_TABLE_DATA(Signal* signal)
  665. {
  666.   jamEntry();
  667.   SubTableData * subTableData = (SubTableData *)signal->getDataPtr();
  668.   SubscriptionRecPtr subRecPtr;
  669.   SubscriptionRecord* subRec;
  670.   subRecPtr.i = subTableData->subscriberData;
  671.   if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
  672.     printf("Trix::execSUB_TABLE_DATA: Failed to find subscription data %un", subRecPtr.i);
  673.     return;
  674.   }
  675.   subRecPtr.p = subRec;
  676.   SegmentedSectionPtr headerPtr, dataPtr;
  677.   if (!signal->getSection(headerPtr, 0)) {
  678.     printf("Trix::execSUB_TABLE_DATA: Failed to get header sectionn");
  679.   }
  680.   if (!signal->getSection(dataPtr, 1)) {
  681.     printf("Trix::execSUB_TABLE_DATA: Failed to get data sectionn");
  682.   }
  683.   executeInsertTransaction(signal, subRecPtr, headerPtr, dataPtr);
  684. }
  686. void Trix::setupSubscription(Signal* signal, SubscriptionRecPtr subRecPtr)
  687. {
  688.   Uint32 attributeList[MAX_ATTRIBUTES_IN_TABLE * 2];
  689.   SubCreateReq * subCreateReq = (SubCreateReq *)signal->getDataPtrSend();
  690.   SubscriptionRecord* subRec = subRecPtr.p;
  691. //  Uint32 listLen = subRec->noOfIndexColumns + subRec->noOfKeyColumns;
  692.   AttrOrderBuffer::DataBufferIterator iter;
  693.   Uint32 i = 0;
  695.   jam();
  696.   bool moreAttributes = subRec->attributeOrder.first(iter);
  697.   while (moreAttributes) {
  698.     attributeList[i++] = *iter.data;
  699.     moreAttributes = subRec->attributeOrder.next(iter);
  700.   }
  701.   // Merge index and key column segments
  702.   struct LinearSectionPtr orderPtr[3];
  703.   orderPtr[0].p = attributeList;
  704.   orderPtr[0].sz = subRec->attributeOrder.getSize();
  707.   subCreateReq->subscriberRef = reference();
  708.   subCreateReq->subscriberData = subRecPtr.i;
  709.   subCreateReq->subscriptionId = subRec->subscriptionId;
  710.   subCreateReq->subscriptionKey = subRec->subscriptionKey;
  711.   subCreateReq->tableId = subRec->sourceTableId;
  712.   subCreateReq->subscriptionType = SubCreateReq::SingleTableScan;
  713.   
  714.   sendSignal(SUMA_REF, GSN_SUB_CREATE_REQ, 
  715.      signal, SubCreateReq::SignalLength+1, JBB, orderPtr, 1);
  716. }
  718. void Trix::setupTableScan(Signal* signal, SubscriptionRecPtr subRecPtr)
  719. {
  720.   SubSyncReq * subSyncReq = (SubSyncReq *)signal->getDataPtrSend();
  722.   jam();
  723.   subSyncReq->subscriptionId = subRecPtr.i;
  724.   subSyncReq->subscriptionKey = subRecPtr.p->subscriptionKey;
  725.   subSyncReq->part = SubscriptionData::MetaData;
  726.   sendSignal(SUMA_REF, GSN_SUB_SYNC_REQ, 
  727.      signal, SubSyncReq::SignalLength, JBB);
  728. }
  730. void Trix::startTableScan(Signal* signal, SubscriptionRecPtr subRecPtr)
  731. {
  732.   jam();
  733.   subRecPtr.p->expectedConf = 1;
  734.   SubSyncReq * subSyncReq = (SubSyncReq *)signal->getDataPtrSend();
  736.   subSyncReq->subscriptionId = subRecPtr.i;
  737.   subSyncReq->subscriptionKey = subRecPtr.p->subscriptionKey;
  738.   subSyncReq->part = SubscriptionData::TableData;
  739.   sendSignal(SUMA_REF, GSN_SUB_SYNC_REQ, 
  740.      signal, SubSyncReq::SignalLength, JBB);
  741. }
  743. void Trix::prepareInsertTransactions(Signal* signal, 
  744.      SubscriptionRecPtr subRecPtr)
  745. {
  746.   SubscriptionRecord* subRec = subRecPtr.p;
  747.   UtilPrepareReq * utilPrepareReq = 
  748.     (UtilPrepareReq *)signal->getDataPtrSend();
  749.   
  750.   jam();
  751.   utilPrepareReq->senderRef = reference();
  752.   utilPrepareReq->senderData = subRecPtr.i;
  754.   const Uint32 pageSizeInWords = 128;
  755.   Uint32 propPage[pageSizeInWords];
  756.   LinearWriter w(&propPage[0],128);
  757.   w.first();
  758.   w.add(UtilPrepareReq::NoOfOperations, 1);
  759.   w.add(UtilPrepareReq::OperationType, UtilPrepareReq::Write);
  760.   w.add(UtilPrepareReq::TableId, subRec->targetTableId);
  761.   // Add index attributes in increasing order and one PK attribute
  762.   for(Uint32 i = 0; i < subRec->noOfIndexColumns + 1; i++)
  763.     w.add(UtilPrepareReq::AttributeId, i);
  765. #if 0
  766.   // Debugging
  767.   SimplePropertiesLinearReader reader(propPage, w.getWordsUsed());
  768.   printf("Trix::prepareInsertTransactions: Sent SimpleProperties:n");
  769.   reader.printAll(ndbout);
  770. #endif
  772.   struct LinearSectionPtr sectionsPtr[UtilPrepareReq::NoOfSections];
  773.   sectionsPtr[UtilPrepareReq::PROPERTIES_SECTION].p = propPage;
  774.   sectionsPtr[UtilPrepareReq::PROPERTIES_SECTION].sz = w.getWordsUsed();
  775.   sendSignal(DBUTIL_REF, GSN_UTIL_PREPARE_REQ, signal,
  776.      UtilPrepareReq::SignalLength, JBB, 
  777.      sectionsPtr, UtilPrepareReq::NoOfSections);
  778. }
  780. void Trix::executeInsertTransaction(Signal* signal, 
  781.     SubscriptionRecPtr subRecPtr,
  782.     SegmentedSectionPtr headerPtr,
  783.     SegmentedSectionPtr dataPtr)
  784. {
  785.   jam();
  786.   SubscriptionRecord* subRec = subRecPtr.p;
  787.   UtilExecuteReq * utilExecuteReq = 
  788.     (UtilExecuteReq *)signal->getDataPtrSend();
  789.   Uint32* headerBuffer = signal->theData + 25;
  790.   Uint32*  dataBuffer = headerBuffer + headerPtr.sz;
  792.   utilExecuteReq->senderRef = reference();
  793.   utilExecuteReq->senderData = subRecPtr.i;
  794.   utilExecuteReq->prepareId = subRec->prepareId;
  795. #if 0
  796.   printf("Header size %un", headerPtr.sz);
  797.   for(int i = 0; i < headerPtr.sz; i++)
  798.     printf("H'%.8x ", headerBuffer[i]);
  799.   printf("n");
  800.   
  801.   printf("Data size %un", dataPtr.sz);
  802.   for(int i = 0; i < dataPtr.sz; i++)
  803.     printf("H'%.8x ", dataBuffer[i]);
  804.   printf("n");
  805. #endif
  806.   // Save scan result in linear buffers
  807.   copy(headerBuffer, headerPtr);
  808.   copy(dataBuffer, dataPtr);
  810.   // Calculate packed key size
  811.   Uint32 noOfKeyData = 0;
  812.   for(Uint32 i = 0; i < headerPtr.sz; i++) {
  813.     AttributeHeader* keyAttrHead = (AttributeHeader *) headerBuffer + i;
  815.     // Filter out NULL attributes
  816.     if (keyAttrHead->isNULL())
  817.       return;
  819.     if (i < subRec->noOfIndexColumns)
  820.       // Renumber index attributes in consequtive order
  821.       keyAttrHead->setAttributeId(i);
  822.     else
  823.       // Calculate total size of PK attribute
  824.       noOfKeyData += keyAttrHead->getDataSize();
  825.   }
  826.   // Increase expected CONF count
  827.   subRec->expectedConf++;
  829.   // Pack key attributes
  830.   AttributeHeader::init(headerBuffer + subRec->noOfIndexColumns,
  831. subRec->noOfIndexColumns,
  832. noOfKeyData);
  834.   struct LinearSectionPtr sectionsPtr[UtilExecuteReq::NoOfSections];
  835.   sectionsPtr[UtilExecuteReq::HEADER_SECTION].p = headerBuffer;
  836.   sectionsPtr[UtilExecuteReq::HEADER_SECTION].sz = 
  837.     subRec->noOfIndexColumns + 1;
  838.   sectionsPtr[UtilExecuteReq::DATA_SECTION].p = dataBuffer;
  839.   sectionsPtr[UtilExecuteReq::DATA_SECTION].sz = dataPtr.sz;
  840.   sendSignal(DBUTIL_REF, GSN_UTIL_EXECUTE_REQ, signal,
  841.      UtilExecuteReq::SignalLength, JBB,
  842.      sectionsPtr, UtilExecuteReq::NoOfSections);
  843. }
  845. void Trix::buildComplete(Signal* signal, SubscriptionRecPtr subRecPtr)
  846. {
  847.   SubRemoveReq * const req = (SubRemoveReq*)signal->getDataPtrSend();
  848.   req->senderRef       = reference();
  849.   req->senderData      = subRecPtr.i;
  850.   req->subscriptionId  = subRecPtr.p->subscriptionId;
  851.   req->subscriptionKey = subRecPtr.p->subscriptionKey;
  852.   sendSignal(SUMA_REF, GSN_SUB_REMOVE_REQ, signal,
  853.      SubRemoveReq::SignalLength, JBB);
  854. }
  856. void Trix::buildFailed(Signal* signal, 
  857.        SubscriptionRecPtr subRecPtr, 
  858.        BuildIndxRef::ErrorCode errorCode)
  859. {
  860.   SubscriptionRecord* subRec = subRecPtr.p;
  862.   subRec->errorCode = errorCode;
  863.   // Continue accumulating since we currently cannot stop SUMA
  864.   subRec->expectedConf--;
  865.   checkParallelism(signal, subRec);
  866.   if (subRec->expectedConf == 0)
  867.     buildComplete(signal, subRecPtr);
  868. }
  870. void
  871. Trix::execSUB_REMOVE_REF(Signal* signal){
  872.   jamEntry();
  873.   //@todo
  874.   ndbrequire(false);
  875. }
  877. void
  878. Trix::execSUB_REMOVE_CONF(Signal* signal){
  879.   jamEntry();
  881.   SubRemoveConf * const conf = (SubRemoveConf*)signal->getDataPtrSend();
  883.   SubscriptionRecPtr subRecPtr;
  884.   c_theSubscriptions.getPtr(subRecPtr, conf->senderData);
  886.   if(subRecPtr.p->prepareId != RNIL){
  887.     jam();
  889.     UtilReleaseReq * const req = (UtilReleaseReq*)signal->getDataPtrSend();
  890.     req->prepareId = subRecPtr.p->prepareId;
  891.     req->senderData = subRecPtr.i;
  893.     sendSignal(DBUTIL_REF, GSN_UTIL_RELEASE_REQ, signal, 
  894.        UtilReleaseReq::SignalLength , JBB);  
  895.     return;
  896.   }
  897.   
  898.   {
  899.     UtilReleaseConf * const conf = (UtilReleaseConf*)signal->getDataPtrSend();
  900.     conf->senderData = subRecPtr.i;
  901.     execUTIL_RELEASE_CONF(signal);
  902.   }
  903. }
  905. void
  906. Trix::execUTIL_RELEASE_REF(Signal* signal){
  907.   jamEntry();
  908.   ndbrequire(false);
  909. }
  911. void
  912. Trix::execUTIL_RELEASE_CONF(Signal* signal){
  913.   
  914.   UtilReleaseConf * const conf = (UtilReleaseConf*)signal->getDataPtrSend();
  915.   
  916.   SubscriptionRecPtr subRecPtr;
  917.   c_theSubscriptions.getPtr(subRecPtr, conf->senderData);
  918.   
  919.   if(subRecPtr.p->errorCode == BuildIndxRef::NoError){
  920.     // Build is complete, reply to original sender
  921.     BuildIndxConf * buildIndxConf = (BuildIndxConf *)signal->getDataPtrSend();
  922.     buildIndxConf->setUserRef(subRecPtr.p->userReference);
  923.     buildIndxConf->setConnectionPtr(subRecPtr.p->connectionPtr);
  924.     buildIndxConf->setRequestType(BuildIndxReq::RT_TRIX);
  925.     buildIndxConf->setIndexType(subRecPtr.p->indexType);
  926.     buildIndxConf->setTableId(subRecPtr.p->sourceTableId);
  927.     buildIndxConf->setIndexId(subRecPtr.p->targetTableId);
  928.     
  929.     sendSignal(subRecPtr.p->userReference, GSN_BUILDINDXCONF, signal, 
  930.        BuildIndxConf::SignalLength , JBB);  
  931.   } else {
  932.     // Build failed, reply to original sender
  933.     BuildIndxRef * buildIndxRef = (BuildIndxRef *)signal->getDataPtrSend();
  934.     buildIndxRef->setUserRef(subRecPtr.p->userReference);
  935.     buildIndxRef->setConnectionPtr(subRecPtr.p->connectionPtr);
  936.     buildIndxRef->setRequestType(BuildIndxReq::RT_TRIX);
  937.     buildIndxRef->setIndexType(subRecPtr.p->indexType);
  938.     buildIndxRef->setTableId(subRecPtr.p->sourceTableId);
  939.     buildIndxRef->setIndexId(subRecPtr.p->targetTableId);
  940.     buildIndxRef->setErrorCode(subRecPtr.p->errorCode);
  941.     
  942.     sendSignal(subRecPtr.p->userReference, GSN_BUILDINDXREF, signal, 
  943.        BuildIndxRef::SignalLength , JBB);  
  944.   }
  946.   // Release subscription record
  947.   subRecPtr.p->attributeOrder.release();
  948.   c_theSubscriptions.release(subRecPtr.i);
  949. }
  951. void Trix::checkParallelism(Signal* signal, SubscriptionRecord* subRec)
  952. {
  953.   if ((subRec->pendingSubSyncContinueConf) &&
  954.       (subRec->expectedConf < subRec->parallelism)) {
  955.     SubSyncContinueConf  * subSyncContinueConf = 
  956.       (SubSyncContinueConf *) signal->getDataPtrSend();
  957.     subSyncContinueConf->subscriptionId = subRec->subscriptionId;
  958.     subSyncContinueConf->subscriptionKey = subRec->subscriptionKey;
  959.     sendSignal(SUMA_REF, GSN_SUB_SYNC_CONTINUE_CONF, signal, 
  960.        SubSyncContinueConf::SignalLength , JBB);  
  961.     subRec->pendingSubSyncContinueConf = false;
  962.   }
  963. }
  965. BLOCK_FUNCTIONS(Trix)
  967. template void append(DataBuffer<15>&,SegmentedSectionPtr,SectionSegmentPool&);