← Back to branch summary

~ubuntu-branches/ubuntu/precise/mysql-5.1/precise

« back to all changes in this revision

Viewing changes to storage/ndb/src/kernel/blocks/dbtux/DbtuxNode.cpp

  • Committer: Bazaar Package Importer
  • Author(s): Norbert Tretkowski
  • Date: 2010-03-17 14:56:02 UTC
  • Revision ID: james.westby@ubuntu.com-20100317145602-x7e30l1b2sb5s6w6
Tags: upstream-5.1.45
Import upstream version 5.1.45

Show diffs side-by-side

added added

removed removed

Lines of Context:
storage/ndb/src/kernel/blocks/dbtux/DbtuxNode.cpp
 
1
/* Copyright (C) 2003 MySQL AB
 
2
 
 
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; version 2 of the License.
 
6
 
 
7
   This program is distributed in the hope that it will be useful,
 
8
   but WITHOUT ANY WARRANTY; without even the implied warranty of
 
9
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
10
   GNU General Public License for more details.
 
11
 
 
12
   You should have received a copy of the GNU General Public License
 
13
   along with this program; if not, write to the Free Software
 
14
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
15
 
 
16
#define DBTUX_NODE_CPP
 
17
#include "Dbtux.hpp"
 
18
 
 
19
/*
 
20
 * Allocate index node in TUP.
 
21
 */
 
22
int
 
23
Dbtux::allocNode(Signal* signal, NodeHandle& node)
 
24
{
 
25
  if (ERROR_INSERTED(12007)) {
 
26
    jam();
 
27
    CLEAR_ERROR_INSERT_VALUE;
 
28
    return TuxMaintReq::NoMemError;
 
29
  }
 
30
  Frag& frag = node.m_frag;
 
31
  Uint32 pageId = NullTupLoc.getPageId();
 
32
  Uint32 pageOffset = NullTupLoc.getPageOffset();
 
33
  Uint32* node32 = 0;
 
34
  int errorCode = c_tup->tuxAllocNode(signal, frag.m_tupIndexFragPtrI, pageId, pageOffset, node32);
 
35
  jamEntry();
 
36
  if (errorCode == 0) {
 
37
    jam();
 
38
    node.m_loc = TupLoc(pageId, pageOffset);
 
39
    node.m_node = reinterpret_cast<TreeNode*>(node32);
 
40
    ndbrequire(node.m_loc != NullTupLoc && node.m_node != 0);
 
41
  } else {
 
42
    switch (errorCode) {
 
43
    case 827:
 
44
      errorCode = TuxMaintReq::NoMemError;
 
45
      break;
 
46
    }
 
47
  }
 
48
  return errorCode;
 
49
}
 
50
 
 
51
/*
 
52
 * Set handle to point to existing node.
 
53
 */
 
54
void
 
55
Dbtux::selectNode(NodeHandle& node, TupLoc loc)
 
56
{
 
57
  Frag& frag = node.m_frag;
 
58
  ndbrequire(loc != NullTupLoc);
 
59
  Uint32 pageId = loc.getPageId();
 
60
  Uint32 pageOffset = loc.getPageOffset();
 
61
  Uint32* node32 = 0;
 
62
  c_tup->tuxGetNode(frag.m_tupIndexFragPtrI, pageId, pageOffset, node32);
 
63
  jamEntry();
 
64
  node.m_loc = loc;
 
65
  node.m_node = reinterpret_cast<TreeNode*>(node32);
 
66
  ndbrequire(node.m_loc != NullTupLoc && node.m_node != 0);
 
67
}
 
68
 
 
69
/*
 
70
 * Set handle to point to new node.  Uses a pre-allocated node.
 
71
 */
 
72
void
 
73
Dbtux::insertNode(NodeHandle& node)
 
74
{
 
75
  Frag& frag = node.m_frag;
 
76
  // unlink from freelist
 
77
  selectNode(node, frag.m_freeLoc);
 
78
  frag.m_freeLoc = node.getLink(0);
 
79
  new (node.m_node) TreeNode();
 
80
#ifdef VM_TRACE
 
81
  TreeHead& tree = frag.m_tree;
 
82
  memset(node.getPref(), DataFillByte, tree.m_prefSize << 2);
 
83
  TreeEnt* entList = tree.getEntList(node.m_node);
 
84
  memset(entList, NodeFillByte, (tree.m_maxOccup + 1) * (TreeEntSize << 2));
 
85
#endif
 
86
}
 
87
 
 
88
/*
 
89
 * Delete existing node.  Simply put it on the freelist.
 
90
 */
 
91
void
 
92
Dbtux::deleteNode(NodeHandle& node)
 
93
{
 
94
  Frag& frag = node.m_frag;
 
95
  ndbrequire(node.getOccup() == 0);
 
96
  // link to freelist
 
97
  node.setLink(0, frag.m_freeLoc);
 
98
  frag.m_freeLoc = node.m_loc;
 
99
  // invalidate the handle
 
100
  node.m_loc = NullTupLoc;
 
101
  node.m_node = 0;
 
102
}
 
103
 
 
104
/*
 
105
 * Set prefix.  Copies the number of words that fits.  Includes
 
106
 * attribute headers for now.  XXX use null mask instead
 
107
 */
 
108
void
 
109
Dbtux::setNodePref(NodeHandle& node)
 
110
{
 
111
  const Frag& frag = node.m_frag;
 
112
  const TreeHead& tree = frag.m_tree;
 
113
  readKeyAttrs(frag, node.getMinMax(0), 0, c_entryKey);
 
114
  copyAttrs(frag, c_entryKey, node.getPref(), tree.m_prefSize);
 
115
}
 
116
 
 
117
// node operations
 
118
 
 
119
/*
 
120
 * Add entry at position.  Move entries greater than or equal to the old
 
121
 * one (if any) to the right.
 
122
 *
 
123
 *            X
 
124
 *            v
 
125
 *      A B C D E _ _  =>  A B C X D E _
 
126
 *      0 1 2 3 4 5 6      0 1 2 3 4 5 6
 
127
 *
 
128
 * Add list of scans at the new entry.
 
129
 */
 
130
void
 
131
Dbtux::nodePushUp(NodeHandle& node, unsigned pos, const TreeEnt& ent, Uint32 scanList)
 
132
{
 
133
  Frag& frag = node.m_frag;
 
134
  TreeHead& tree = frag.m_tree;
 
135
  const unsigned occup = node.getOccup();
 
136
  ndbrequire(occup < tree.m_maxOccup && pos <= occup);
 
137
  // fix old scans
 
138
  if (node.getNodeScan() != RNIL)
 
139
    nodePushUpScans(node, pos);
 
140
  // fix node
 
141
  TreeEnt* const entList = tree.getEntList(node.m_node);
 
142
  entList[occup] = entList[0];
 
143
  TreeEnt* const tmpList = entList + 1;
 
144
  for (unsigned i = occup; i > pos; i--) {
 
145
    jam();
 
146
    tmpList[i] = tmpList[i - 1];
 
147
  }
 
148
  tmpList[pos] = ent;
 
149
  entList[0] = entList[occup + 1];
 
150
  node.setOccup(occup + 1);
 
151
  // add new scans
 
152
  if (scanList != RNIL)
 
153
    addScanList(node, pos, scanList);
 
154
  // fix prefix
 
155
  if (occup == 0 || pos == 0)
 
156
    setNodePref(node);
 
157
}
 
158
 
 
159
void
 
160
Dbtux::nodePushUpScans(NodeHandle& node, unsigned pos)
 
161
{
 
162
  const unsigned occup = node.getOccup();
 
163
  ScanOpPtr scanPtr;
 
164
  scanPtr.i = node.getNodeScan();
 
165
  do {
 
166
    jam();
 
167
    c_scanOpPool.getPtr(scanPtr);
 
168
    TreePos& scanPos = scanPtr.p->m_scanPos;
 
169
    ndbrequire(scanPos.m_loc == node.m_loc && scanPos.m_pos < occup);
 
170
    if (scanPos.m_pos >= pos) {
 
171
      jam();
 
172
#ifdef VM_TRACE
 
173
      if (debugFlags & DebugScan) {
 
174
        debugOut << "Fix scan " << scanPtr.i << " " << *scanPtr.p << endl;
 
175
        debugOut << "At pushUp pos=" << pos << " " << node << endl;
 
176
      }
 
177
#endif
 
178
      scanPos.m_pos++;
 
179
    }
 
180
    scanPtr.i = scanPtr.p->m_nodeScan;
 
181
  } while (scanPtr.i != RNIL);
 
182
}
 
183
 
 
184
/*
 
185
 * Remove and return entry at position.  Move entries greater than the
 
186
 * removed one to the left.  This is the opposite of nodePushUp.
 
187
 *
 
188
 *                               D
 
189
 *            ^                  ^
 
190
 *      A B C D E F _  =>  A B C E F _ _
 
191
 *      0 1 2 3 4 5 6      0 1 2 3 4 5 6
 
192
 *
 
193
 * Scans at removed entry are returned if non-zero location is passed or
 
194
 * else moved forward.
 
195
 */
 
196
void
 
197
Dbtux::nodePopDown(NodeHandle& node, unsigned pos, TreeEnt& ent, Uint32* scanList)
 
198
{
 
199
  Frag& frag = node.m_frag;
 
200
  TreeHead& tree = frag.m_tree;
 
201
  const unsigned occup = node.getOccup();
 
202
  ndbrequire(occup <= tree.m_maxOccup && pos < occup);
 
203
  if (node.getNodeScan() != RNIL) {
 
204
    // remove or move scans at this position
 
205
    if (scanList == 0)
 
206
      moveScanList(node, pos);
 
207
    else
 
208
      removeScanList(node, pos, *scanList);
 
209
    // fix other scans
 
210
    if (node.getNodeScan() != RNIL)
 
211
      nodePopDownScans(node, pos);
 
212
  }
 
213
  // fix node
 
214
  TreeEnt* const entList = tree.getEntList(node.m_node);
 
215
  entList[occup] = entList[0];
 
216
  TreeEnt* const tmpList = entList + 1;
 
217
  ent = tmpList[pos];
 
218
  for (unsigned i = pos; i < occup - 1; i++) {
 
219
    jam();
 
220
    tmpList[i] = tmpList[i + 1];
 
221
  }
 
222
  entList[0] = entList[occup - 1];
 
223
  node.setOccup(occup - 1);
 
224
  // fix prefix
 
225
  if (occup != 1 && pos == 0)
 
226
    setNodePref(node);
 
227
}
 
228
 
 
229
void
 
230
Dbtux::nodePopDownScans(NodeHandle& node, unsigned pos)
 
231
{
 
232
  const unsigned occup = node.getOccup();
 
233
  ScanOpPtr scanPtr;
 
234
  scanPtr.i = node.getNodeScan();
 
235
  do {
 
236
    jam();
 
237
    c_scanOpPool.getPtr(scanPtr);
 
238
    TreePos& scanPos = scanPtr.p->m_scanPos;
 
239
    ndbrequire(scanPos.m_loc == node.m_loc && scanPos.m_pos < occup);
 
240
    // handled before
 
241
    ndbrequire(scanPos.m_pos != pos);
 
242
    if (scanPos.m_pos > pos) {
 
243
      jam();
 
244
#ifdef VM_TRACE
 
245
      if (debugFlags & DebugScan) {
 
246
        debugOut << "Fix scan " << scanPtr.i << " " << *scanPtr.p << endl;
 
247
        debugOut << "At popDown pos=" << pos << " " << node << endl;
 
248
      }
 
249
#endif
 
250
      scanPos.m_pos--;
 
251
    }
 
252
    scanPtr.i = scanPtr.p->m_nodeScan;
 
253
  } while (scanPtr.i != RNIL);
 
254
}
 
255
 
 
256
/*
 
257
 * Add entry at existing position.  Move entries less than or equal to
 
258
 * the old one to the left.  Remove and return old min entry.
 
259
 *
 
260
 *            X            A
 
261
 *      ^     v            ^
 
262
 *      A B C D E _ _  =>  B C D X E _ _
 
263
 *      0 1 2 3 4 5 6      0 1 2 3 4 5 6
 
264
 *
 
265
 * Return list of scans at the removed position 0.
 
266
 */
 
267
void
 
268
Dbtux::nodePushDown(NodeHandle& node, unsigned pos, TreeEnt& ent, Uint32& scanList)
 
269
{
 
270
  Frag& frag = node.m_frag;
 
271
  TreeHead& tree = frag.m_tree;
 
272
  const unsigned occup = node.getOccup();
 
273
  ndbrequire(occup <= tree.m_maxOccup && pos < occup);
 
274
  if (node.getNodeScan() != RNIL) {
 
275
    // remove scans at 0
 
276
    removeScanList(node, 0, scanList);
 
277
    // fix other scans
 
278
    if (node.getNodeScan() != RNIL)
 
279
      nodePushDownScans(node, pos);
 
280
  }
 
281
  // fix node
 
282
  TreeEnt* const entList = tree.getEntList(node.m_node);
 
283
  entList[occup] = entList[0];
 
284
  TreeEnt* const tmpList = entList + 1;
 
285
  TreeEnt oldMin = tmpList[0];
 
286
  for (unsigned i = 0; i < pos; i++) {
 
287
    jam();
 
288
    tmpList[i] = tmpList[i + 1];
 
289
  }
 
290
  tmpList[pos] = ent;
 
291
  ent = oldMin;
 
292
  entList[0] = entList[occup];
 
293
  // fix prefix
 
294
  if (true)
 
295
    setNodePref(node);
 
296
}
 
297
 
 
298
void
 
299
Dbtux::nodePushDownScans(NodeHandle& node, unsigned pos)
 
300
{
 
301
  const unsigned occup = node.getOccup();
 
302
  ScanOpPtr scanPtr;
 
303
  scanPtr.i = node.getNodeScan();
 
304
  do {
 
305
    jam();
 
306
    c_scanOpPool.getPtr(scanPtr);
 
307
    TreePos& scanPos = scanPtr.p->m_scanPos;
 
308
    ndbrequire(scanPos.m_loc == node.m_loc && scanPos.m_pos < occup);
 
309
    // handled before
 
310
    ndbrequire(scanPos.m_pos != 0);
 
311
    if (scanPos.m_pos <= pos) {
 
312
      jam();
 
313
#ifdef VM_TRACE
 
314
      if (debugFlags & DebugScan) {
 
315
        debugOut << "Fix scan " << scanPtr.i << " " << *scanPtr.p << endl;
 
316
        debugOut << "At pushDown pos=" << pos << " " << node << endl;
 
317
      }
 
318
#endif
 
319
      scanPos.m_pos--;
 
320
    }
 
321
    scanPtr.i = scanPtr.p->m_nodeScan;
 
322
  } while (scanPtr.i != RNIL);
 
323
}
 
324
 
 
325
/*
 
326
 * Remove and return entry at position.  Move entries less than the
 
327
 * removed one to the right.  Replace min entry by the input entry.
 
328
 * This is the opposite of nodePushDown.
 
329
 *
 
330
 *      X                        D
 
331
 *      v     ^                  ^
 
332
 *      A B C D E _ _  =>  X A B C E _ _
 
333
 *      0 1 2 3 4 5 6      0 1 2 3 4 5 6
 
334
 *
 
335
 * Move scans at removed entry and add scans at the new entry.
 
336
 */
 
337
void
 
338
Dbtux::nodePopUp(NodeHandle& node, unsigned pos, TreeEnt& ent, Uint32 scanList)
 
339
{
 
340
  Frag& frag = node.m_frag;
 
341
  TreeHead& tree = frag.m_tree;
 
342
  const unsigned occup = node.getOccup();
 
343
  ndbrequire(occup <= tree.m_maxOccup && pos < occup);
 
344
  if (node.getNodeScan() != RNIL) {
 
345
    // move scans whose entry disappears
 
346
    moveScanList(node, pos);
 
347
    // fix other scans
 
348
    if (node.getNodeScan() != RNIL)
 
349
      nodePopUpScans(node, pos);
 
350
  }
 
351
  // fix node
 
352
  TreeEnt* const entList = tree.getEntList(node.m_node);
 
353
  entList[occup] = entList[0];
 
354
  TreeEnt* const tmpList = entList + 1;
 
355
  TreeEnt newMin = ent;
 
356
  ent = tmpList[pos];
 
357
  for (unsigned i = pos; i > 0; i--) {
 
358
    jam();
 
359
    tmpList[i] = tmpList[i - 1];
 
360
  }
 
361
  tmpList[0] = newMin;
 
362
  entList[0] = entList[occup];
 
363
  // add scans
 
364
  if (scanList != RNIL)
 
365
    addScanList(node, 0, scanList);
 
366
  // fix prefix
 
367
  if (true)
 
368
    setNodePref(node);
 
369
}
 
370
 
 
371
void
 
372
Dbtux::nodePopUpScans(NodeHandle& node, unsigned pos)
 
373
{
 
374
  const unsigned occup = node.getOccup();
 
375
  ScanOpPtr scanPtr;
 
376
  scanPtr.i = node.getNodeScan();
 
377
  do {
 
378
    jam();
 
379
    c_scanOpPool.getPtr(scanPtr);
 
380
    TreePos& scanPos = scanPtr.p->m_scanPos;
 
381
    ndbrequire(scanPos.m_loc == node.m_loc && scanPos.m_pos < occup);
 
382
    ndbrequire(scanPos.m_pos != pos);
 
383
    if (scanPos.m_pos < pos) {
 
384
      jam();
 
385
#ifdef VM_TRACE
 
386
      if (debugFlags & DebugScan) {
 
387
        debugOut << "Fix scan " << scanPtr.i << " " << *scanPtr.p << endl;
 
388
        debugOut << "At popUp pos=" << pos << " " << node << endl;
 
389
      }
 
390
#endif
 
391
      scanPos.m_pos++;
 
392
    }
 
393
    scanPtr.i = scanPtr.p->m_nodeScan;
 
394
  } while (scanPtr.i != RNIL);
 
395
}
 
396
 
 
397
/*
 
398
 * Move number of entries from another node to this node before the min
 
399
 * (i=0) or after the max (i=1).  Expensive but not often used.
 
400
 */
 
401
void
 
402
Dbtux::nodeSlide(NodeHandle& dstNode, NodeHandle& srcNode, unsigned cnt, unsigned i)
 
403
{
 
404
  ndbrequire(i <= 1);
 
405
  while (cnt != 0) {
 
406
    TreeEnt ent;
 
407
    Uint32 scanList = RNIL;
 
408
    nodePopDown(srcNode, i == 0 ? srcNode.getOccup() - 1 : 0, ent, &scanList);
 
409
    nodePushUp(dstNode, i == 0 ? 0 : dstNode.getOccup(), ent, scanList);
 
410
    cnt--;
 
411
  }
 
412
}
 
413
 
 
414
// scans linked to node
 
415
 
 
416
 
 
417
/*
 
418
 * Add list of scans to node at given position.
 
419
 */
 
420
void
 
421
Dbtux::addScanList(NodeHandle& node, unsigned pos, Uint32 scanList)
 
422
{
 
423
  ScanOpPtr scanPtr;
 
424
  scanPtr.i = scanList;
 
425
  do {
 
426
    jam();
 
427
    c_scanOpPool.getPtr(scanPtr);
 
428
#ifdef VM_TRACE
 
429
      if (debugFlags & DebugScan) {
 
430
        debugOut << "Add scan " << scanPtr.i << " " << *scanPtr.p << endl;
 
431
        debugOut << "To pos=" << pos << " " << node << endl;
 
432
      }
 
433
#endif
 
434
    const Uint32 nextPtrI = scanPtr.p->m_nodeScan;
 
435
    scanPtr.p->m_nodeScan = RNIL;
 
436
    linkScan(node, scanPtr);
 
437
    TreePos& scanPos = scanPtr.p->m_scanPos;
 
438
    // set position but leave direction alone
 
439
    scanPos.m_loc = node.m_loc;
 
440
    scanPos.m_pos = pos;
 
441
    scanPtr.i = nextPtrI;
 
442
  } while (scanPtr.i != RNIL);
 
443
}
 
444
 
 
445
/*
 
446
 * Remove list of scans from node at given position.  The return
 
447
 * location must point to existing list (in fact RNIL always).
 
448
 */
 
449
void
 
450
Dbtux::removeScanList(NodeHandle& node, unsigned pos, Uint32& scanList)
 
451
{
 
452
  ScanOpPtr scanPtr;
 
453
  scanPtr.i = node.getNodeScan();
 
454
  do {
 
455
    jam();
 
456
    c_scanOpPool.getPtr(scanPtr);
 
457
    const Uint32 nextPtrI = scanPtr.p->m_nodeScan;
 
458
    TreePos& scanPos = scanPtr.p->m_scanPos;
 
459
    ndbrequire(scanPos.m_loc == node.m_loc);
 
460
    if (scanPos.m_pos == pos) {
 
461
      jam();
 
462
#ifdef VM_TRACE
 
463
      if (debugFlags & DebugScan) {
 
464
        debugOut << "Remove scan " << scanPtr.i << " " << *scanPtr.p << endl;
 
465
        debugOut << "Fron pos=" << pos << " " << node << endl;
 
466
      }
 
467
#endif
 
468
      unlinkScan(node, scanPtr);
 
469
      scanPtr.p->m_nodeScan = scanList;
 
470
      scanList = scanPtr.i;
 
471
      // unset position but leave direction alone
 
472
      scanPos.m_loc = NullTupLoc;
 
473
      scanPos.m_pos = ZNIL;
 
474
    }
 
475
    scanPtr.i = nextPtrI;
 
476
  } while (scanPtr.i != RNIL);
 
477
}
 
478
 
 
479
/*
 
480
 * Move list of scans away from entry about to be removed.  Uses scan
 
481
 * method scanNext().
 
482
 */
 
483
void
 
484
Dbtux::moveScanList(NodeHandle& node, unsigned pos)
 
485
{
 
486
  ScanOpPtr scanPtr;
 
487
  scanPtr.i = node.getNodeScan();
 
488
  do {
 
489
    jam();
 
490
    c_scanOpPool.getPtr(scanPtr);
 
491
    TreePos& scanPos = scanPtr.p->m_scanPos;
 
492
    const Uint32 nextPtrI = scanPtr.p->m_nodeScan;
 
493
    ndbrequire(scanPos.m_loc == node.m_loc);
 
494
    if (scanPos.m_pos == pos) {
 
495
      jam();
 
496
#ifdef VM_TRACE
 
497
      if (debugFlags & DebugScan) {
 
498
        debugOut << "Move scan " << scanPtr.i << " " << *scanPtr.p << endl;
 
499
        debugOut << "At pos=" << pos << " " << node << endl;
 
500
      }
 
501
#endif
 
502
      scanNext(scanPtr, true);
 
503
      ndbrequire(! (scanPos.m_loc == node.m_loc && scanPos.m_pos == pos));
 
504
    }
 
505
    scanPtr.i = nextPtrI;
 
506
  } while (scanPtr.i != RNIL);
 
507
}
 
508
 
 
509
/*
 
510
 * Link scan to the list under the node.  The list is single-linked and
 
511
 * ordering does not matter.
 
512
 */
 
513
void
 
514
Dbtux::linkScan(NodeHandle& node, ScanOpPtr scanPtr)
 
515
{
 
516
#ifdef VM_TRACE
 
517
  if (debugFlags & DebugScan) {
 
518
    debugOut << "Link scan " << scanPtr.i << " " << *scanPtr.p << endl;
 
519
    debugOut << "To node " << node << endl;
 
520
  }
 
521
#endif
 
522
  ndbrequire(! islinkScan(node, scanPtr) && scanPtr.p->m_nodeScan == RNIL);
 
523
  scanPtr.p->m_nodeScan = node.getNodeScan();
 
524
  node.setNodeScan(scanPtr.i);
 
525
}
 
526
 
 
527
/*
 
528
 * Unlink a scan from the list under the node.
 
529
 */
 
530
void
 
531
Dbtux::unlinkScan(NodeHandle& node, ScanOpPtr scanPtr)
 
532
{
 
533
#ifdef VM_TRACE
 
534
  if (debugFlags & DebugScan) {
 
535
    debugOut << "Unlink scan " << scanPtr.i << " " << *scanPtr.p << endl;
 
536
    debugOut << "From node " << node << endl;
 
537
  }
 
538
#endif
 
539
  ScanOpPtr currPtr;
 
540
  currPtr.i = node.getNodeScan();
 
541
  ScanOpPtr prevPtr;
 
542
  prevPtr.i = RNIL;
 
543
  while (true) {
 
544
    jam();
 
545
    c_scanOpPool.getPtr(currPtr);
 
546
    Uint32 nextPtrI = currPtr.p->m_nodeScan;
 
547
    if (currPtr.i == scanPtr.i) {
 
548
      jam();
 
549
      if (prevPtr.i == RNIL) {
 
550
        node.setNodeScan(nextPtrI);
 
551
      } else {
 
552
        jam();
 
553
        prevPtr.p->m_nodeScan = nextPtrI;
 
554
      }
 
555
      scanPtr.p->m_nodeScan = RNIL;
 
556
      // check for duplicates
 
557
      ndbrequire(! islinkScan(node, scanPtr));
 
558
      return;
 
559
    }
 
560
    prevPtr = currPtr;
 
561
    currPtr.i = nextPtrI;
 
562
  }
 
563
}
 
564
 
 
565
/*
 
566
 * Check if a scan is linked to this node.  Only for ndbrequire.
 
567
 */
 
568
bool
 
569
Dbtux::islinkScan(NodeHandle& node, ScanOpPtr scanPtr)
 
570
{
 
571
  ScanOpPtr currPtr;
 
572
  currPtr.i = node.getNodeScan();
 
573
  while (currPtr.i != RNIL) {
 
574
    jam();
 
575
    c_scanOpPool.getPtr(currPtr);
 
576
    if (currPtr.i == scanPtr.i) {
 
577
      jam();
 
578
      return true;
 
579
    }
 
580
    currPtr.i = currPtr.p->m_nodeScan;
 
581
  }
 
582
  return false;
 
583
}
 
584
 
 
585
void
 
586
Dbtux::NodeHandle::progError(int line, int cause, const char* file)
 
587
{
 
588
  ErrorReporter::handleAssert("Dbtux::NodeHandle: assert failed", file, line);
 
589
}