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

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

MySQL数据库

开发平台：

Visual C++

DbtuxScan.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 DBTUX_SCAN_CPP
  18. #include "Dbtux.hpp"
  20. void
  21. Dbtux::execACC_SCANREQ(Signal* signal)
  22. {
  23.   jamEntry();
  24.   const AccScanReq reqCopy = *(const AccScanReq*)signal->getDataPtr();
  25.   const AccScanReq* const req = &reqCopy;
  26.   ScanOpPtr scanPtr;
  27.   scanPtr.i = RNIL;
  28.   do {
  29.     // get the index
  30.     IndexPtr indexPtr;
  31.     c_indexPool.getPtr(indexPtr, req->tableId);
  32.     // get the fragment
  33.     FragPtr fragPtr;
  34.     fragPtr.i = RNIL;
  35.     for (unsigned i = 0; i < indexPtr.p->m_numFrags; i++) {
  36.       jam();
  37.       if (indexPtr.p->m_fragId[i] == req->fragmentNo) {
  38.         jam();
  39.         c_fragPool.getPtr(fragPtr, indexPtr.p->m_fragPtrI[i]);
  40.         break;
  41.       }
  42.     }
  43.     ndbrequire(fragPtr.i != RNIL);
  44.     Frag& frag = *fragPtr.p;
  45.     // must be normal DIH/TC fragment
  46.     ndbrequire(frag.m_fragId < (1 << frag.m_fragOff));
  47.     TreeHead& tree = frag.m_tree;
  48.     // check for empty fragment
  49.     if (tree.m_root == NullTupLoc) {
  50.       jam();
  51.       AccScanConf* const conf = (AccScanConf*)signal->getDataPtrSend();
  52.       conf->scanPtr = req->senderData;
  53.       conf->accPtr = RNIL;
  54.       conf->flag = AccScanConf::ZEMPTY_FRAGMENT;
  55.       sendSignal(req->senderRef, GSN_ACC_SCANCONF,
  56.           signal, AccScanConf::SignalLength, JBB);
  57.       return;
  58.     }
  59.     // seize from pool and link to per-fragment list
  60.     if (! frag.m_scanList.seize(scanPtr)) {
  61.       jam();
  62.       break;
  63.     }
  64.     new (scanPtr.p) ScanOp(c_scanBoundPool);
  65.     scanPtr.p->m_state = ScanOp::First;
  66.     scanPtr.p->m_userPtr = req->senderData;
  67.     scanPtr.p->m_userRef = req->senderRef;
  68.     scanPtr.p->m_tableId = indexPtr.p->m_tableId;
  69.     scanPtr.p->m_indexId = indexPtr.i;
  70.     scanPtr.p->m_fragId = fragPtr.p->m_fragId;
  71.     scanPtr.p->m_fragPtrI = fragPtr.i;
  72.     scanPtr.p->m_transId1 = req->transId1;
  73.     scanPtr.p->m_transId2 = req->transId2;
  74.     scanPtr.p->m_savePointId = req->savePointId;
  75.     scanPtr.p->m_readCommitted = AccScanReq::getReadCommittedFlag(req->requestInfo);
  76.     scanPtr.p->m_lockMode = AccScanReq::getLockMode(req->requestInfo);
  77.     scanPtr.p->m_keyInfo = AccScanReq::getKeyinfoFlag(req->requestInfo);
  78. #ifdef VM_TRACE
  79.     if (debugFlags & DebugScan) {
  80.       debugOut << "Seize scan " << scanPtr.i << " " << *scanPtr.p << endl;
  81.     }
  82. #endif
  83.     /*
  84.      * readCommitted lockMode keyInfo
  85.      * 1 0 0 - read committed (no lock)
  86.      * 0 0 0 - read latest (read lock)
  87.      * 0 1 1 - read exclusive (write lock)
  88.      */
  89.     // conf
  90.     AccScanConf* const conf = (AccScanConf*)signal->getDataPtrSend();
  91.     conf->scanPtr = req->senderData;
  92.     conf->accPtr = scanPtr.i;
  93.     conf->flag = AccScanConf::ZNOT_EMPTY_FRAGMENT;
  94.     sendSignal(req->senderRef, GSN_ACC_SCANCONF,
  95.         signal, AccScanConf::SignalLength, JBB);
  96.     return;
  97.   } while (0);
  98.   if (scanPtr.i != RNIL) {
  99.     jam();
  100.     releaseScanOp(scanPtr);
  101.   }
  102.   // LQH does not handle REF
  103.   signal->theData[0] = 0x313;
  104.   sendSignal(req->senderRef, GSN_ACC_SCANREF,
  105.       signal, 1, JBB);
  106. }
  108. /*
  109.  * Receive bounds for scan in single direct call.  The bounds can arrive
  110.  * in any order.  Attribute ids are those of index table.
  111.  *
  112.  * Replace EQ by equivalent LE + GE.  Check for conflicting bounds.
  113.  * Check that sets of lower and upper bounds are on initial sequences of
  114.  * keys and that all but possibly last bound is non-strict.
  115.  *
  116.  * Finally save the sets of lower and upper bounds (i.e. start key and
  117.  * end key).  Full bound type (< 4) is included but only the strict bit
  118.  * is used since lower and upper have now been separated.
  119.  */
  120. void
  121. Dbtux::execTUX_BOUND_INFO(Signal* signal)
  122. {
  123.   jamEntry();
  124.   struct BoundInfo {
  125.     int type;
  126.     unsigned offset;
  127.     unsigned size;
  128.   };
  129.   TuxBoundInfo* const sig = (TuxBoundInfo*)signal->getDataPtrSend();
  130.   const TuxBoundInfo reqCopy = *(const TuxBoundInfo*)sig;
  131.   const TuxBoundInfo* const req = &reqCopy;
  132.   // get records
  133.   ScanOp& scan = *c_scanOpPool.getPtr(req->tuxScanPtrI);
  134.   Index& index = *c_indexPool.getPtr(scan.m_indexId);
  135.   // collect lower and upper bounds
  136.   BoundInfo boundInfo[2][MaxIndexAttributes];
  137.   // largest attrId seen plus one
  138.   Uint32 maxAttrId[2] = { 0, 0 };
  139.   unsigned offset = 0;
  140.   const Uint32* const data = (Uint32*)sig + TuxBoundInfo::SignalLength;
  141.   // walk through entries
  142.   while (offset + 2 <= req->boundAiLength) {
  143.     jam();
  144.     const unsigned type = data[offset];
  145.     if (type > 4) {
  146.       jam();
  147.       scan.m_state = ScanOp::Invalid;
  148.       sig->errorCode = TuxBoundInfo::InvalidAttrInfo;
  149.       return;
  150.     }
  151.     const AttributeHeader* ah = (const AttributeHeader*)&data[offset + 1];
  152.     const Uint32 attrId = ah->getAttributeId();
  153.     const Uint32 dataSize = ah->getDataSize();
  154.     if (attrId >= index.m_numAttrs) {
  155.       jam();
  156.       scan.m_state = ScanOp::Invalid;
  157.       sig->errorCode = TuxBoundInfo::InvalidAttrInfo;
  158.       return;
  159.     }
  160.     for (unsigned j = 0; j <= 1; j++) {
  161.       // check if lower/upper bit matches
  162.       const unsigned luBit = (j << 1);
  163.       if ((type & 0x2) != luBit && type != 4)
  164.         continue;
  165.       // EQ -> LE, GE
  166.       const unsigned type2 = (type & 0x1) | luBit;
  167.       // fill in any gap
  168.       while (maxAttrId[j] <= attrId) {
  169.         BoundInfo& b = boundInfo[j][maxAttrId[j]++];
  170.         b.type = -1;
  171.       }
  172.       BoundInfo& b = boundInfo[j][attrId];
  173.       if (b.type != -1) {
  174.         // compare with previous bound
  175.         if (b.type != (int)type2 ||
  176.             b.size != 2 + dataSize ||
  177.             memcmp(&data[b.offset + 2], &data[offset + 2], dataSize << 2) != 0) {
  178.           jam();
  179.           scan.m_state = ScanOp::Invalid;
  180.           sig->errorCode = TuxBoundInfo::InvalidBounds;
  181.           return;
  182.         }
  183.       } else {
  184.         // enter new bound
  185.         b.type = type2;
  186.         b.offset = offset;
  187.         b.size = 2 + dataSize;
  188.       }
  189.     }
  190.     // jump to next
  191.     offset += 2 + dataSize;
  192.   }
  193.   if (offset != req->boundAiLength) {
  194.     jam();
  195.     scan.m_state = ScanOp::Invalid;
  196.     sig->errorCode = TuxBoundInfo::InvalidAttrInfo;
  197.     return;
  198.   }
  199.   for (unsigned j = 0; j <= 1; j++) {
  200.     // save lower/upper bound in index attribute id order
  201.     for (unsigned i = 0; i < maxAttrId[j]; i++) {
  202.       jam();
  203.       const BoundInfo& b = boundInfo[j][i];
  204.       // check for gap or strict bound before last
  205.       if (b.type == -1 || (i + 1 < maxAttrId[j] && (b.type & 0x1))) {
  206.         jam();
  207.         scan.m_state = ScanOp::Invalid;
  208.         sig->errorCode = TuxBoundInfo::InvalidBounds;
  209.         return;
  210.       }
  211.       bool ok = scan.m_bound[j]->append(&data[b.offset], b.size);
  212.       if (! ok) {
  213.         jam();
  214.         scan.m_state = ScanOp::Invalid;
  215.         sig->errorCode = TuxBoundInfo::OutOfBuffers;
  216.         return;
  217.       }
  218.     }
  219.     scan.m_boundCnt[j] = maxAttrId[j];
  220.   }
  221.   // no error
  222.   sig->errorCode = 0;
  223. }
  225. void
  226. Dbtux::execNEXT_SCANREQ(Signal* signal)
  227. {
  228.   jamEntry();
  229.   const NextScanReq reqCopy = *(const NextScanReq*)signal->getDataPtr();
  230.   const NextScanReq* const req = &reqCopy;
  231.   ScanOpPtr scanPtr;
  232.   scanPtr.i = req->accPtr;
  233.   c_scanOpPool.getPtr(scanPtr);
  234.   ScanOp& scan = *scanPtr.p;
  235.   Frag& frag = *c_fragPool.getPtr(scan.m_fragPtrI);
  236. #ifdef VM_TRACE
  237.   if (debugFlags & DebugScan) {
  238.     debugOut << "NEXT_SCANREQ scan " << scanPtr.i << " " << scan << endl;
  239.   }
  240. #endif
  241.   // handle unlock previous and close scan
  242.   switch (req->scanFlag) {
  243.   case NextScanReq::ZSCAN_NEXT:
  244.     jam();
  245.     break;
  246.   case NextScanReq::ZSCAN_NEXT_COMMIT:
  247.     jam();
  248.   case NextScanReq::ZSCAN_COMMIT:
  249.     jam();
  250.     if (! scan.m_readCommitted) {
  251.       jam();
  252.       AccLockReq* const lockReq = (AccLockReq*)signal->getDataPtrSend();
  253.       lockReq->returnCode = RNIL;
  254.       lockReq->requestInfo = AccLockReq::Unlock;
  255.       lockReq->accOpPtr = req->accOperationPtr;
  256.       EXECUTE_DIRECT(DBACC, GSN_ACC_LOCKREQ, signal, AccLockReq::UndoSignalLength);
  257.       jamEntry();
  258.       ndbrequire(lockReq->returnCode == AccLockReq::Success);
  259.       removeAccLockOp(scan, req->accOperationPtr);
  260.     }
  261.     if (req->scanFlag == NextScanReq::ZSCAN_COMMIT) {
  262.       jam();
  263.       NextScanConf* const conf = (NextScanConf*)signal->getDataPtrSend();
  264.       conf->scanPtr = scan.m_userPtr;
  265.       unsigned signalLength = 1;
  266.       sendSignal(scanPtr.p->m_userRef, GSN_NEXT_SCANCONF,
  267.           signal, signalLength, JBB);
  268.       return;
  269.     }
  270.     break;
  271.   case NextScanReq::ZSCAN_CLOSE:
  272.     jam();
  273.     // unlink from tree node first to avoid state changes
  274.     if (scan.m_scanPos.m_loc != NullTupLoc) {
  275.       jam();
  276.       const TupLoc loc = scan.m_scanPos.m_loc;
  277.       NodeHandle node(frag);
  278.       selectNode(node, loc);
  279.       unlinkScan(node, scanPtr);
  280.       scan.m_scanPos.m_loc = NullTupLoc;
  281.     }
  282.     if (scan.m_lockwait) {
  283.       jam();
  284.       ndbrequire(scan.m_accLockOp != RNIL);
  285.       // use ACC_ABORTCONF to flush out any reply in job buffer
  286.       AccLockReq* const lockReq = (AccLockReq*)signal->getDataPtrSend();
  287.       lockReq->returnCode = RNIL;
  288.       lockReq->requestInfo = AccLockReq::AbortWithConf;
  289.       lockReq->accOpPtr = scan.m_accLockOp;
  290.       EXECUTE_DIRECT(DBACC, GSN_ACC_LOCKREQ, signal, AccLockReq::UndoSignalLength);
  291.       jamEntry();
  292.       ndbrequire(lockReq->returnCode == AccLockReq::Success);
  293.       scan.m_state = ScanOp::Aborting;
  294.       return;
  295.     }
  296.     if (scan.m_state == ScanOp::Locked) {
  297.       jam();
  298.       ndbrequire(scan.m_accLockOp != RNIL);
  299.       AccLockReq* const lockReq = (AccLockReq*)signal->getDataPtrSend();
  300.       lockReq->returnCode = RNIL;
  301.       lockReq->requestInfo = AccLockReq::Unlock;
  302.       lockReq->accOpPtr = scan.m_accLockOp;
  303.       EXECUTE_DIRECT(DBACC, GSN_ACC_LOCKREQ, signal, AccLockReq::UndoSignalLength);
  304.       jamEntry();
  305.       ndbrequire(lockReq->returnCode == AccLockReq::Success);
  306.       scan.m_accLockOp = RNIL;
  307.     }
  308.     scan.m_state = ScanOp::Aborting;
  309.     scanClose(signal, scanPtr);
  310.     return;
  311.   case NextScanReq::ZSCAN_NEXT_ABORT:
  312.     jam();
  313.   default:
  314.     jam();
  315.     ndbrequire(false);
  316.     break;
  317.   }
  318.   // start looking for next scan result
  319.   AccCheckScan* checkReq = (AccCheckScan*)signal->getDataPtrSend();
  320.   checkReq->accPtr = scanPtr.i;
  321.   checkReq->checkLcpStop = AccCheckScan::ZNOT_CHECK_LCP_STOP;
  322.   EXECUTE_DIRECT(DBTUX, GSN_ACC_CHECK_SCAN, signal, AccCheckScan::SignalLength);
  323.   jamEntry();
  324. }
  326. void
  327. Dbtux::execACC_CHECK_SCAN(Signal* signal)
  328. {
  329.   jamEntry();
  330.   const AccCheckScan reqCopy = *(const AccCheckScan*)signal->getDataPtr();
  331.   const AccCheckScan* const req = &reqCopy;
  332.   ScanOpPtr scanPtr;
  333.   scanPtr.i = req->accPtr;
  334.   c_scanOpPool.getPtr(scanPtr);
  335.   ScanOp& scan = *scanPtr.p;
  336.   Frag& frag = *c_fragPool.getPtr(scan.m_fragPtrI);
  337. #ifdef VM_TRACE
  338.   if (debugFlags & DebugScan) {
  339.     debugOut << "ACC_CHECK_SCAN scan " << scanPtr.i << " " << scan << endl;
  340.   }
  341. #endif
  342.   if (req->checkLcpStop == AccCheckScan::ZCHECK_LCP_STOP) {
  343.     jam();
  344.     signal->theData[0] = scan.m_userPtr;
  345.     signal->theData[1] = true;
  346.     EXECUTE_DIRECT(DBLQH, GSN_CHECK_LCP_STOP, signal, 2);
  347.     jamEntry();
  348.     return;   // stop
  349.   }
  350.   if (scan.m_lockwait) {
  351.     jam();
  352.     // LQH asks if we are waiting for lock and we tell it to ask again
  353.     const TreeEnt ent = scan.m_scanEnt;
  354.     NextScanConf* const conf = (NextScanConf*)signal->getDataPtrSend();
  355.     conf->scanPtr = scan.m_userPtr;
  356.     conf->accOperationPtr = RNIL;       // no tuple returned
  357.     conf->fragId = frag.m_fragId | (ent.m_fragBit << frag.m_fragOff);
  358.     unsigned signalLength = 3;
  359.     // if TC has ordered scan close, it will be detected here
  360.     sendSignal(scan.m_userRef, GSN_NEXT_SCANCONF,
  361.         signal, signalLength, JBB);
  362.     return;     // stop
  363.   }
  364.   if (scan.m_state == ScanOp::First) {
  365.     jam();
  366.     // search is done only once in single range scan
  367.     scanFirst(scanPtr);
  368. #ifdef VM_TRACE
  369.     if (debugFlags & DebugScan) {
  370.       debugOut << "First scan " << scanPtr.i << " " << scan << endl;
  371.     }
  372. #endif
  373.   }
  374.   if (scan.m_state == ScanOp::Next) {
  375.     jam();
  376.     // look for next
  377.     scanNext(scanPtr);
  378.   }
  379.   // for reading tuple key in Current or Locked state
  380.   Data pkData = c_dataBuffer;
  381.   unsigned pkSize = 0; // indicates not yet done
  382.   if (scan.m_state == ScanOp::Current) {
  383.     // found an entry to return
  384.     jam();
  385.     ndbrequire(scan.m_accLockOp == RNIL);
  386.     if (! scan.m_readCommitted) {
  387.       jam();
  388.       const TreeEnt ent = scan.m_scanEnt;
  389.       // read tuple key
  390.       readTablePk(frag, ent, pkData, pkSize);
  391.       // get read lock or exclusive lock
  392.       AccLockReq* const lockReq = (AccLockReq*)signal->getDataPtrSend();
  393.       lockReq->returnCode = RNIL;
  394.       lockReq->requestInfo =
  395.         scan.m_lockMode == 0 ? AccLockReq::LockShared : AccLockReq::LockExclusive;
  396.       lockReq->accOpPtr = RNIL;
  397.       lockReq->userPtr = scanPtr.i;
  398.       lockReq->userRef = reference();
  399.       lockReq->tableId = scan.m_tableId;
  400.       lockReq->fragId = frag.m_fragId | (ent.m_fragBit << frag.m_fragOff);
  401.       lockReq->fragPtrI = frag.m_accTableFragPtrI[ent.m_fragBit];
  402.       const Uint32* const buf32 = static_cast<Uint32*>(pkData);
  403.       const Uint64* const buf64 = reinterpret_cast<const Uint64*>(buf32);
  404.       lockReq->hashValue = md5_hash(buf64, pkSize);
  405.       lockReq->tupAddr = getTupAddr(frag, ent);
  406.       lockReq->transId1 = scan.m_transId1;
  407.       lockReq->transId2 = scan.m_transId2;
  408.       // execute
  409.       EXECUTE_DIRECT(DBACC, GSN_ACC_LOCKREQ, signal, AccLockReq::LockSignalLength);
  410.       jamEntry();
  411.       switch (lockReq->returnCode) {
  412.       case AccLockReq::Success:
  413.         jam();
  414.         scan.m_state = ScanOp::Locked;
  415.         scan.m_accLockOp = lockReq->accOpPtr;
  416. #ifdef VM_TRACE
  417.         if (debugFlags & DebugScan) {
  418.           debugOut << "Lock immediate scan " << scanPtr.i << " " << scan << endl;
  419.         }
  420. #endif
  421.         break;
  422.       case AccLockReq::IsBlocked:
  423.         jam();
  424.         // normal lock wait
  425.         scan.m_state = ScanOp::Blocked;
  426.         scan.m_lockwait = true;
  427.         scan.m_accLockOp = lockReq->accOpPtr;
  428. #ifdef VM_TRACE
  429.         if (debugFlags & DebugScan) {
  430.           debugOut << "Lock wait scan " << scanPtr.i << " " << scan << endl;
  431.         }
  432. #endif
  433.         // LQH will wake us up
  434.         signal->theData[0] = scan.m_userPtr;
  435.         signal->theData[1] = true;
  436.         EXECUTE_DIRECT(DBLQH, GSN_CHECK_LCP_STOP, signal, 2);
  437.         jamEntry();
  438.         return;  // stop
  439.         break;
  440.       case AccLockReq::Refused:
  441.         jam();
  442.         // we cannot see deleted tuple (assert only)
  443.         ndbassert(false);
  444.         // skip it
  445.         scan.m_state = ScanOp::Next;
  446.         signal->theData[0] = scan.m_userPtr;
  447.         signal->theData[1] = true;
  448.         EXECUTE_DIRECT(DBLQH, GSN_CHECK_LCP_STOP, signal, 2);
  449.         jamEntry();
  450.         return;  // stop
  451.         break;
  452.       case AccLockReq::NoFreeOp:
  453.         jam();
  454.         // max ops should depend on max scans (assert only)
  455.         ndbassert(false);
  456.         // stay in Current state
  457.         scan.m_state = ScanOp::Current;
  458.         signal->theData[0] = scan.m_userPtr;
  459.         signal->theData[1] = true;
  460.         EXECUTE_DIRECT(DBLQH, GSN_CHECK_LCP_STOP, signal, 2);
  461.         jamEntry();
  462.         return;  // stop
  463.         break;
  464.       default:
  465.         ndbrequire(false);
  466.         break;
  467.       }
  468.     } else {
  469.       scan.m_state = ScanOp::Locked;
  470.     }
  471.   }
  472.   if (scan.m_state == ScanOp::Locked) {
  473.     // we have lock or do not need one
  474.     jam();
  475.     // read keys if not already done (uses signal)
  476.     const TreeEnt ent = scan.m_scanEnt;
  477.     if (scan.m_keyInfo) {
  478.       jam();
  479.       if (pkSize == 0) {
  480.         jam();
  481.         readTablePk(frag, ent, pkData, pkSize);
  482.       }
  483.     }
  484.     // conf signal
  485.     NextScanConf* const conf = (NextScanConf*)signal->getDataPtrSend();
  486.     conf->scanPtr = scan.m_userPtr;
  487.     // the lock is passed to LQH
  488.     Uint32 accLockOp = scan.m_accLockOp;
  489.     if (accLockOp != RNIL) {
  490.       scan.m_accLockOp = RNIL;
  491.       // remember it until LQH unlocks it
  492.       addAccLockOp(scan, accLockOp);
  493.     } else {
  494.       ndbrequire(scan.m_readCommitted);
  495.       // operation RNIL in LQH would signal no tuple returned
  496.       accLockOp = (Uint32)-1;
  497.     }
  498.     conf->accOperationPtr = accLockOp;
  499.     conf->fragId = frag.m_fragId | (ent.m_fragBit << frag.m_fragOff);
  500.     conf->localKey[0] = getTupAddr(frag, ent);
  501.     conf->localKey[1] = 0;
  502.     conf->localKeyLength = 1;
  503.     unsigned signalLength = 6;
  504.     // add key info
  505.     if (scan.m_keyInfo) {
  506.       jam();
  507.       conf->keyLength = pkSize;
  508.       // piggy-back first 4 words of key data
  509.       for (unsigned i = 0; i < 4; i++) {
  510.         conf->key[i] = i < pkSize ? pkData[i] : 0;
  511.       }
  512.       signalLength = 11;
  513.     }
  514.     if (! scan.m_readCommitted) {
  515.       sendSignal(scan.m_userRef, GSN_NEXT_SCANCONF,
  516.           signal, signalLength, JBB);
  517.     } else {
  518.       Uint32 blockNo = refToBlock(scan.m_userRef);
  519.       EXECUTE_DIRECT(blockNo, GSN_NEXT_SCANCONF, signal, signalLength);
  520.     }
  521.     // send rest of key data
  522.     if (scan.m_keyInfo && pkSize > 4) {
  523.       unsigned total = 4;
  524.       while (total < pkSize) {
  525.         jam();
  526.         unsigned length = pkSize - total;
  527.         if (length > 20)
  528.           length = 20;
  529.         signal->theData[0] = scan.m_userPtr;
  530.         signal->theData[1] = 0;
  531.         signal->theData[2] = 0;
  532.         signal->theData[3] = length;
  533.         memcpy(&signal->theData[4], &pkData[total], length << 2);
  534.         sendSignal(scan.m_userRef, GSN_ACC_SCAN_INFO24,
  535.             signal, 4 + length, JBB);
  536.         total += length;
  537.       }
  538.     }
  539.     // next time look for next entry
  540.     scan.m_state = ScanOp::Next;
  541.     return;
  542.   }
  543.   // XXX in ACC this is checked before req->checkLcpStop
  544.   if (scan.m_state == ScanOp::Last ||
  545.       scan.m_state == ScanOp::Invalid) {
  546.     jam();
  547.     NextScanConf* const conf = (NextScanConf*)signal->getDataPtrSend();
  548.     conf->scanPtr = scan.m_userPtr;
  549.     conf->accOperationPtr = RNIL;
  550.     conf->fragId = RNIL;
  551.     unsigned signalLength = 3;
  552.     sendSignal(scanPtr.p->m_userRef, GSN_NEXT_SCANCONF,
  553.         signal, signalLength, JBB);
  554.     return;
  555.   }
  556.   ndbrequire(false);
  557. }
  559. /*
  560.  * Lock succeeded (after delay) in ACC.  If the lock is for current
  561.  * entry, set state to Locked.  If the lock is for an entry we were
  562.  * moved away from, simply unlock it.  Finally, if we are closing the
  563.  * scan, do nothing since we have already sent an abort request.
  564.  */
  565. void
  566. Dbtux::execACCKEYCONF(Signal* signal)
  567. {
  568.   jamEntry();
  569.   ScanOpPtr scanPtr;
  570.   scanPtr.i = signal->theData[0];
  571.   c_scanOpPool.getPtr(scanPtr);
  572.   ScanOp& scan = *scanPtr.p;
  573. #ifdef VM_TRACE
  574.   if (debugFlags & DebugScan) {
  575.     debugOut << "Lock obtained scan " << scanPtr.i << " " << scan << endl;
  576.   }
  577. #endif
  578.   ndbrequire(scan.m_lockwait && scan.m_accLockOp != RNIL);
  579.   scan.m_lockwait = false;
  580.   if (scan.m_state == ScanOp::Blocked) {
  581.     // the lock wait was for current entry
  582.     jam();
  583.     scan.m_state = ScanOp::Locked;
  584.     // LQH has the ball
  585.     return;
  586.   }
  587.   if (scan.m_state != ScanOp::Aborting) {
  588.     // we were moved, release lock
  589.     jam();
  590.     AccLockReq* const lockReq = (AccLockReq*)signal->getDataPtrSend();
  591.     lockReq->returnCode = RNIL;
  592.     lockReq->requestInfo = AccLockReq::Unlock;
  593.     lockReq->accOpPtr = scan.m_accLockOp;
  594.     EXECUTE_DIRECT(DBACC, GSN_ACC_LOCKREQ, signal, AccLockReq::UndoSignalLength);
  595.     jamEntry();
  596.     ndbrequire(lockReq->returnCode == AccLockReq::Success);
  597.     scan.m_accLockOp = RNIL;
  598.     // LQH has the ball
  599.     return;
  600.   }
  601.   // lose the lock
  602.   scan.m_accLockOp = RNIL;
  603.   // continue at ACC_ABORTCONF
  604. }
  606. /*
  607.  * Lock failed (after delay) in ACC.  Probably means somebody ahead of
  608.  * us in lock queue deleted the tuple.
  609.  */
  610. void
  611. Dbtux::execACCKEYREF(Signal* signal)
  612. {
  613.   jamEntry();
  614.   ScanOpPtr scanPtr;
  615.   scanPtr.i = signal->theData[0];
  616.   c_scanOpPool.getPtr(scanPtr);
  617.   ScanOp& scan = *scanPtr.p;
  618. #ifdef VM_TRACE
  619.   if (debugFlags & DebugScan) {
  620.     debugOut << "Lock refused scan " << scanPtr.i << " " << scan << endl;
  621.   }
  622. #endif
  623.   ndbrequire(scan.m_lockwait && scan.m_accLockOp != RNIL);
  624.   scan.m_lockwait = false;
  625.   if (scan.m_state != ScanOp::Aborting) {
  626.     jam();
  627.     // release the operation
  628.     AccLockReq* const lockReq = (AccLockReq*)signal->getDataPtrSend();
  629.     lockReq->returnCode = RNIL;
  630.     lockReq->requestInfo = AccLockReq::Abort;
  631.     lockReq->accOpPtr = scan.m_accLockOp;
  632.     EXECUTE_DIRECT(DBACC, GSN_ACC_LOCKREQ, signal, AccLockReq::UndoSignalLength);
  633.     jamEntry();
  634.     ndbrequire(lockReq->returnCode == AccLockReq::Success);
  635.     scan.m_accLockOp = RNIL;
  636.     // scan position should already have been moved (assert only)
  637.     if (scan.m_state == ScanOp::Blocked) {
  638.       jam();
  639.       ndbassert(false);
  640.       scan.m_state = ScanOp::Next;
  641.     }
  642.     // LQH has the ball
  643.     return;
  644.   }
  645.   // lose the lock
  646.   scan.m_accLockOp = RNIL;
  647.   // continue at ACC_ABORTCONF
  648. }
  650. /*
  651.  * Received when scan is closing.  This signal arrives after any
  652.  * ACCKEYCON or ACCKEYREF which may have been in job buffer.
  653.  */
  654. void
  655. Dbtux::execACC_ABORTCONF(Signal* signal)
  656. {
  657.   jamEntry();
  658.   ScanOpPtr scanPtr;
  659.   scanPtr.i = signal->theData[0];
  660.   c_scanOpPool.getPtr(scanPtr);
  661.   ScanOp& scan = *scanPtr.p;
  662. #ifdef VM_TRACE
  663.   if (debugFlags & DebugScan) {
  664.     debugOut << "ACC_ABORTCONF scan " << scanPtr.i << " " << scan << endl;
  665.   }
  666. #endif
  667.   ndbrequire(scan.m_state == ScanOp::Aborting);
  668.   // most likely we are still in lock wait
  669.   if (scan.m_lockwait) {
  670.     jam();
  671.     scan.m_lockwait = false;
  672.     scan.m_accLockOp = RNIL;
  673.   }
  674.   scanClose(signal, scanPtr);
  675. }
  677. /*
  678.  * Find start position for single range scan.  If it exists, sets state
  679.  * to Next and links the scan to the node.  The first entry is returned
  680.  * by scanNext.
  681.  */
  682. void
  683. Dbtux::scanFirst(ScanOpPtr scanPtr)
  684. {
  685.   ScanOp& scan = *scanPtr.p;
  686.   Frag& frag = *c_fragPool.getPtr(scan.m_fragPtrI);
  687.   TreeHead& tree = frag.m_tree;
  688.   // set up index keys for this operation
  689.   setKeyAttrs(frag);
  690.   // unpack lower bound into c_dataBuffer
  691.   const ScanBound& bound = *scan.m_bound[0];
  692.   ScanBoundIterator iter;
  693.   bound.first(iter);
  694.   for (unsigned j = 0; j < bound.getSize(); j++) {
  695.     jam();
  696.     c_dataBuffer[j] = *iter.data;
  697.     bound.next(iter);
  698.   }
  699.   // search for scan start position
  700.   TreePos treePos;
  701.   searchToScan(frag, c_dataBuffer, scan.m_boundCnt[0], treePos);
  702.   if (treePos.m_loc == NullTupLoc) {
  703.     // empty tree
  704.     jam();
  705.     scan.m_state = ScanOp::Last;
  706.     return;
  707.   }
  708.   // set position and state
  709.   scan.m_scanPos = treePos;
  710.   scan.m_state = ScanOp::Next;
  711.   // link the scan to node found
  712.   NodeHandle node(frag);
  713.   selectNode(node, treePos.m_loc);
  714.   linkScan(node, scanPtr);
  715. }
  717. /*
  718.  * Move to next entry.  The scan is already linked to some node.  When
  719.  * we leave, if an entry was found, it will be linked to a possibly
  720.  * different node.  The scan has a position, and a direction which tells
  721.  * from where we came to this position.  This is one of:
  722.  *
  723.  * 0 - up from left child (scan this node next)
  724.  * 1 - up from right child (proceed to parent)
  725.  * 2 - up from root (the scan ends)
  726.  * 3 - left to right within node (at end proceed to right child)
  727.  * 4 - down from parent (proceed to left child)
  728.  *
  729.  * If an entry was found, scan direction is 3.  Therefore tree
  730.  * re-organizations need not worry about scan direction.
  731.  */
  732. void
  733. Dbtux::scanNext(ScanOpPtr scanPtr)
  734. {
  735.   ScanOp& scan = *scanPtr.p;
  736.   Frag& frag = *c_fragPool.getPtr(scan.m_fragPtrI);
  737. #ifdef VM_TRACE
  738.   if (debugFlags & DebugScan) {
  739.     debugOut << "Next in scan " << scanPtr.i << " " << scan << endl;
  740.   }
  741. #endif
  742.   // cannot be moved away from tuple we have locked
  743.   ndbrequire(scan.m_state != ScanOp::Locked);
  744.   // set up index keys for this operation
  745.   setKeyAttrs(frag);
  746.   // unpack upper bound into c_dataBuffer
  747.   const ScanBound& bound = *scan.m_bound[1];
  748.   ScanBoundIterator iter;
  749.   bound.first(iter);
  750.   for (unsigned j = 0; j < bound.getSize(); j++) {
  751.     jam();
  752.     c_dataBuffer[j] = *iter.data;
  753.     bound.next(iter);
  754.   }
  755.   // use copy of position
  756.   TreePos pos = scan.m_scanPos;
  757.   // get and remember original node
  758.   NodeHandle origNode(frag);
  759.   selectNode(origNode, pos.m_loc);
  760.   ndbrequire(islinkScan(origNode, scanPtr));
  761.   // current node in loop
  762.   NodeHandle node = origNode;
  763.   // copy of entry found
  764.   TreeEnt ent;
  765.   while (true) {
  766.     jam();
  767.     if (pos.m_dir == 2) {
  768.       // coming up from root ends the scan
  769.       jam();
  770.       pos.m_loc = NullTupLoc;
  771.       scan.m_state = ScanOp::Last;
  772.       break;
  773.     }
  774.     if (node.m_loc != pos.m_loc) {
  775.       jam();
  776.       selectNode(node, pos.m_loc);
  777.     }
  778.     if (pos.m_dir == 4) {
  779.       // coming down from parent proceed to left child
  780.       jam();
  781.       TupLoc loc = node.getLink(0);
  782.       if (loc != NullTupLoc) {
  783.         jam();
  784.         pos.m_loc = loc;
  785.         pos.m_dir = 4;  // unchanged
  786.         continue;
  787.       }
  788.       // pretend we came from left child
  789.       pos.m_dir = 0;
  790.     }
  791.     if (pos.m_dir == 0) {
  792.       // coming up from left child scan current node
  793.       jam();
  794.       pos.m_pos = 0;
  795.       pos.m_match = false;
  796.       pos.m_dir = 3;
  797.     }
  798.     if (pos.m_dir == 3) {
  799.       // within node
  800.       jam();
  801.       unsigned occup = node.getOccup();
  802.       ndbrequire(occup >= 1);
  803.       // advance position
  804.       if (! pos.m_match)
  805.         pos.m_match = true;
  806.       else
  807.         pos.m_pos++;
  808.       if (pos.m_pos < occup) {
  809.         jam();
  810.         ent = node.getEnt(pos.m_pos);
  811.         pos.m_dir = 3;  // unchanged
  812.         // read and compare all attributes
  813.         readKeyAttrs(frag, ent, 0, c_entryKey);
  814.         int ret = cmpScanBound(frag, 1, c_dataBuffer, scan.m_boundCnt[1], c_entryKey);
  815.         ndbrequire(ret != NdbSqlUtil::CmpUnknown);
  816.         if (ret < 0) {
  817.           jam();
  818.           // hit upper bound of single range scan
  819.           pos.m_loc = NullTupLoc;
  820.           scan.m_state = ScanOp::Last;
  821.           break;
  822.         }
  823.         // can we see it
  824.         if (! scanVisible(scanPtr, ent)) {
  825.           jam();
  826.           continue;
  827.         }
  828.         // found entry
  829.         scan.m_state = ScanOp::Current;
  830.         break;
  831.       }
  832.       // after node proceed to right child
  833.       TupLoc loc = node.getLink(1);
  834.       if (loc != NullTupLoc) {
  835.         jam();
  836.         pos.m_loc = loc;
  837.         pos.m_dir = 4;
  838.         continue;
  839.       }
  840.       // pretend we came from right child
  841.       pos.m_dir = 1;
  842.     }
  843.     if (pos.m_dir == 1) {
  844.       // coming up from right child proceed to parent
  845.       jam();
  846.       pos.m_loc = node.getLink(2);
  847.       pos.m_dir = node.getSide();
  848.       continue;
  849.     }
  850.     ndbrequire(false);
  851.   }
  852.   // copy back position
  853.   scan.m_scanPos = pos;
  854.   // relink
  855.   if (scan.m_state == ScanOp::Current) {
  856.     ndbrequire(pos.m_match == true && pos.m_dir == 3);
  857.     ndbrequire(pos.m_loc == node.m_loc);
  858.     if (origNode.m_loc != node.m_loc) {
  859.       jam();
  860.       unlinkScan(origNode, scanPtr);
  861.       linkScan(node, scanPtr);
  862.     }
  863.     // copy found entry
  864.     scan.m_scanEnt = ent;
  865.   } else if (scan.m_state == ScanOp::Last) {
  866.     jam();
  867.     ndbrequire(pos.m_loc == NullTupLoc);
  868.     unlinkScan(origNode, scanPtr);
  869.   } else {
  870.     ndbrequire(false);
  871.   }
  872. #ifdef VM_TRACE
  873.   if (debugFlags & DebugScan) {
  874.     debugOut << "Next out scan " << scanPtr.i << " " << scan << endl;
  875.   }
  876. #endif
  877. }
  879. /*
  880.  * Check if an entry is visible to the scan.
  881.  *
  882.  * There is a special check to never accept same tuple twice in a row.
  883.  * This is faster than asking TUP.  It also fixes some special cases
  884.  * which are not analyzed or handled yet.
  885.  */
  886. bool
  887. Dbtux::scanVisible(ScanOpPtr scanPtr, TreeEnt ent)
  888. {
  889.   const ScanOp& scan = *scanPtr.p;
  890.   const Frag& frag = *c_fragPool.getPtr(scan.m_fragPtrI);
  891.   Uint32 fragBit = ent.m_fragBit;
  892.   Uint32 tableFragPtrI = frag.m_tupTableFragPtrI[fragBit];
  893.   Uint32 fragId = frag.m_fragId | (fragBit << frag.m_fragOff);
  894.   Uint32 tupAddr = getTupAddr(frag, ent);
  895.   Uint32 tupVersion = ent.m_tupVersion;
  896.   // check for same tuple twice in row
  897.   if (scan.m_scanEnt.m_tupLoc == ent.m_tupLoc &&
  898.       scan.m_scanEnt.m_fragBit == fragBit) {
  899.     jam();
  900.     return false;
  901.   }
  902.   Uint32 transId1 = scan.m_transId1;
  903.   Uint32 transId2 = scan.m_transId2;
  904.   Uint32 savePointId = scan.m_savePointId;
  905.   bool ret = c_tup->tuxQueryTh(tableFragPtrI, tupAddr, tupVersion, transId1, transId2, savePointId);
  906.   jamEntry();
  907.   return ret;
  908. }
  910. /*
  911.  * Finish closing of scan and send conf.  Any lock wait has been done
  912.  * already.
  913.  */
  914. void
  915. Dbtux::scanClose(Signal* signal, ScanOpPtr scanPtr)
  916. {
  917.   ScanOp& scan = *scanPtr.p;
  918.   ndbrequire(! scan.m_lockwait && scan.m_accLockOp == RNIL);
  919.   // unlock all not unlocked by LQH
  920.   for (unsigned i = 0; i < scan.m_maxAccLockOps; i++) {
  921.     if (scan.m_accLockOps[i] != RNIL) {
  922.       jam();
  923.       AccLockReq* const lockReq = (AccLockReq*)signal->getDataPtrSend();
  924.       lockReq->returnCode = RNIL;
  925.       lockReq->requestInfo = AccLockReq::Abort;
  926.       lockReq->accOpPtr = scan.m_accLockOps[i];
  927.       EXECUTE_DIRECT(DBACC, GSN_ACC_LOCKREQ, signal, AccLockReq::UndoSignalLength);
  928.       jamEntry();
  929.       ndbrequire(lockReq->returnCode == AccLockReq::Success);
  930.       scan.m_accLockOps[i] = RNIL;
  931.     }
  932.   }
  933.   // send conf
  934.   NextScanConf* const conf = (NextScanConf*)signal->getDataPtrSend();
  935.   conf->scanPtr = scanPtr.p->m_userPtr;
  936.   conf->accOperationPtr = RNIL;
  937.   conf->fragId = RNIL;
  938.   unsigned signalLength = 3;
  939.   sendSignal(scanPtr.p->m_userRef, GSN_NEXT_SCANCONF,
  940.       signal, signalLength, JBB);
  941.   releaseScanOp(scanPtr);
  942. }
  944. void
  945. Dbtux::addAccLockOp(ScanOp& scan, Uint32 accLockOp)
  946. {
  947.   ndbrequire(accLockOp != RNIL);
  948.   Uint32* list = scan.m_accLockOps;
  949.   bool ok = false;
  950.   for (unsigned i = 0; i < scan.m_maxAccLockOps; i++) {
  951.     ndbrequire(list[i] != accLockOp);
  952.     if (! ok && list[i] == RNIL) {
  953.       list[i] = accLockOp;
  954.       ok = true;
  955.       // continue check for duplicates
  956.     }
  957.   }
  958.   if (! ok) {
  959.     unsigned i = scan.m_maxAccLockOps;
  960.     if (i < MaxAccLockOps) {
  961.       list[i] = accLockOp;
  962.       ok = true;
  963.       scan.m_maxAccLockOps = i + 1;
  964.     }
  965.   }
  966.   ndbrequire(ok);
  967. }
  969. void
  970. Dbtux::removeAccLockOp(ScanOp& scan, Uint32 accLockOp)
  971. {
  972.   ndbrequire(accLockOp != RNIL);
  973.   Uint32* list = scan.m_accLockOps;
  974.   bool ok = false;
  975.   for (unsigned i = 0; i < scan.m_maxAccLockOps; i++) {
  976.     if (list[i] == accLockOp) {
  977.       list[i] = RNIL;
  978.       ok = true;
  979.       break;
  980.     }
  981.   }
  982.   ndbrequire(ok);
  983. }
  985. /*
  986.  * Release allocated records.
  987.  */
  988. void
  989. Dbtux::releaseScanOp(ScanOpPtr& scanPtr)
  990. {
  991. #ifdef VM_TRACE
  992.   if (debugFlags & DebugScan) {
  993.     debugOut << "Release scan " << scanPtr.i << " " << *scanPtr.p << endl;
  994.   }
  995. #endif
  996.   Frag& frag = *c_fragPool.getPtr(scanPtr.p->m_fragPtrI);
  997.   scanPtr.p->m_boundMin.release();
  998.   scanPtr.p->m_boundMax.release();
  999.   // unlink from per-fragment list and release from pool
  1000.   frag.m_scanList.release(scanPtr);
  1001. }