MySQL 5.6.14 Source Code Document
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
DbtupCommit.cpp
1 /*
2  Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
3 
4  This program is free software; you can redistribute it and/or modify
5  it under the terms of the GNU General Public License as published by
6  the Free Software Foundation; version 2 of the License.
7 
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.
12 
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., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
16 */
17 
18 #define DBTUP_C
19 #define DBTUP_COMMIT_CPP
20 #include "Dbtup.hpp"
21 #include <RefConvert.hpp>
22 #include <ndb_limits.h>
23 #include <pc.hpp>
24 #include <signaldata/TupCommit.hpp>
25 #include "../dblqh/Dblqh.hpp"
26 
27 void Dbtup::execTUP_DEALLOCREQ(Signal* signal)
28 {
29  TablerecPtr regTabPtr;
30  FragrecordPtr regFragPtr;
31  Uint32 frag_page_id, frag_id;
32 
33  jamEntry();
34 
35  frag_id= signal->theData[0];
36  regTabPtr.i= signal->theData[1];
37  frag_page_id= signal->theData[2];
38  Uint32 page_index= signal->theData[3];
39 
40  ptrCheckGuard(regTabPtr, cnoOfTablerec, tablerec);
41 
42  getFragmentrec(regFragPtr, frag_id, regTabPtr.p);
43  ndbassert(regFragPtr.p != NULL);
44 
45  if (! Local_key::isInvalid(frag_page_id, page_index))
46  {
47  Local_key tmp;
48  tmp.m_page_no= getRealpid(regFragPtr.p, frag_page_id);
49  tmp.m_page_idx= page_index;
50 
51  PagePtr pagePtr;
52  Tuple_header* ptr= (Tuple_header*)get_ptr(&pagePtr, &tmp, regTabPtr.p);
53 
54  ndbassert(ptr->m_header_bits & Tuple_header::FREED);
55 
56  if (regTabPtr.p->m_attributes[MM].m_no_of_varsize +
57  regTabPtr.p->m_attributes[MM].m_no_of_dynamic)
58  {
59  jam();
60  free_var_rec(regFragPtr.p, regTabPtr.p, &tmp, pagePtr);
61  } else {
62  free_fix_rec(regFragPtr.p, regTabPtr.p, &tmp, (Fix_page*)pagePtr.p);
63  }
64  }
65 }
66 
67 void Dbtup::execTUP_WRITELOG_REQ(Signal* signal)
68 {
69  jamEntry();
70  OperationrecPtr loopOpPtr;
71  loopOpPtr.i= signal->theData[0];
72  Uint32 gci_hi = signal->theData[1];
73  Uint32 gci_lo = signal->theData[2];
74  c_operation_pool.getPtr(loopOpPtr);
75  while (loopOpPtr.p->prevActiveOp != RNIL) {
76  jam();
77  loopOpPtr.i= loopOpPtr.p->prevActiveOp;
78  c_operation_pool.getPtr(loopOpPtr);
79  }
80  do {
81  ndbrequire(get_trans_state(loopOpPtr.p) == TRANS_STARTED);
82  signal->theData[0] = loopOpPtr.p->userpointer;
83  signal->theData[1] = gci_hi;
84  signal->theData[2] = gci_lo;
85  if (loopOpPtr.p->nextActiveOp == RNIL) {
86  jam();
87  EXECUTE_DIRECT(DBLQH, GSN_LQH_WRITELOG_REQ, signal, 3);
88  return;
89  }
90  jam();
91  EXECUTE_DIRECT(DBLQH, GSN_LQH_WRITELOG_REQ, signal, 3);
92  jamEntry();
93  loopOpPtr.i= loopOpPtr.p->nextActiveOp;
94  c_operation_pool.getPtr(loopOpPtr);
95  } while (true);
96 }
97 
98 /* ---------------------------------------------------------------- */
99 /* INITIALIZATION OF ONE CONNECTION RECORD TO PREPARE FOR NEXT OP. */
100 /* ---------------------------------------------------------------- */
101 void Dbtup::initOpConnection(Operationrec* regOperPtr)
102 {
103  set_tuple_state(regOperPtr, TUPLE_ALREADY_ABORTED);
104  set_trans_state(regOperPtr, TRANS_IDLE);
105  regOperPtr->op_struct.op_type= ZREAD;
106  regOperPtr->op_struct.m_disk_preallocated= 0;
107  regOperPtr->op_struct.m_load_diskpage_on_commit= 0;
108  regOperPtr->op_struct.m_wait_log_buffer= 0;
109  regOperPtr->op_struct.in_active_list = false;
110  regOperPtr->m_undo_buffer_space= 0;
111 }
112 
113 bool
114 Dbtup::is_rowid_lcp_scanned(const Local_key& key1,
115  const Dbtup::ScanOp& op)
116 {
117  Local_key key2 = op.m_scanPos.m_key;
118  switch (op.m_state) {
119  case Dbtup::ScanOp::First:
120  ndbrequire(key2.isNull());
121  return false;
122  case Dbtup::ScanOp::Current:
123  case Dbtup::ScanOp::Next:
124  ndbrequire(!key2.isNull());
125  if (key1.m_page_no < key2.m_page_no)
126  return true;
127  if (key1.m_page_no > key2.m_page_no)
128  return false;
129  if (key1.m_page_idx < key2.m_page_idx)
130  return true;
131  if (key1.m_page_idx > key2.m_page_idx)
132  return false;
136  if (op.m_state == Dbtup::ScanOp::Next)
137  return true;
138  return false;
139  case Dbtup::ScanOp::Last:
140  return true;
141  default:
142  ndbrequire(false);
143  break;
144  }
145  return false;
146 }
147 
148 void
149 Dbtup::dealloc_tuple(Signal* signal,
150  Uint32 gci_hi,
151  Uint32 gci_lo,
152  Page* page,
153  Tuple_header* ptr,
154  KeyReqStruct * req_struct,
155  Operationrec* regOperPtr,
156  Fragrecord* regFragPtr,
157  Tablerec* regTabPtr)
158 {
159  Uint32 lcpScan_ptr_i= regFragPtr->m_lcp_scan_op;
160 
161  Uint32 bits = ptr->m_header_bits;
162  Uint32 extra_bits = Tuple_header::FREED;
163  if (bits & Tuple_header::DISK_PART)
164  {
165  jam();
166  Local_key disk;
167  memcpy(&disk, ptr->get_disk_ref_ptr(regTabPtr), sizeof(disk));
168  PagePtr tmpptr;
169  tmpptr.i = m_pgman_ptr.i;
170  tmpptr.p = reinterpret_cast<Page*>(m_pgman_ptr.p);
171  disk_page_free(signal, regTabPtr, regFragPtr,
172  &disk, tmpptr, gci_hi);
173  }
174 
175  if (! (bits & (Tuple_header::LCP_SKIP | Tuple_header::ALLOC)) &&
176  lcpScan_ptr_i != RNIL && regTabPtr->m_no_of_disk_attributes > 0)
177  {
178  jam();
179  ScanOpPtr scanOp;
180  c_scanOpPool.getPtr(scanOp, lcpScan_ptr_i);
181  Local_key rowid = regOperPtr->m_tuple_location;
182  rowid.m_page_no = page->frag_page_id;
183  if (!is_rowid_lcp_scanned(rowid, *scanOp.p))
184  {
185  jam();
186 
193  handle_lcp_keep_commit(&rowid,
194  req_struct, regOperPtr, regFragPtr, regTabPtr);
195  }
196  }
197 
198  ptr->m_header_bits = bits | extra_bits;
199 
200  if (regTabPtr->m_bits & Tablerec::TR_RowGCI)
201  {
202  jam();
203  * ptr->get_mm_gci(regTabPtr) = gci_hi;
204  if (regTabPtr->m_bits & Tablerec::TR_ExtraRowGCIBits)
205  {
206  Uint32 attrId = regTabPtr->getExtraAttrId<Tablerec::TR_ExtraRowGCIBits>();
207  store_extra_row_bits(attrId, regTabPtr, ptr, gci_lo, /* truncate */true);
208  }
209  }
210 }
211 
212 void
213 Dbtup::handle_lcp_keep_commit(const Local_key* rowid,
214  KeyReqStruct * req_struct,
215  Operationrec * opPtrP,
216  Fragrecord * regFragPtr,
217  Tablerec * regTabPtr)
218 {
219  bool disk = false;
220  Uint32 sizes[4];
221  Uint32 * copytuple = get_copy_tuple_raw(&opPtrP->m_copy_tuple_location);
222  Tuple_header * dst = get_copy_tuple(copytuple);
223  Tuple_header * org = req_struct->m_tuple_ptr;
224  if (regTabPtr->need_expand(disk))
225  {
226  setup_fixed_part(req_struct, opPtrP, regTabPtr);
227  req_struct->m_tuple_ptr = dst;
228  expand_tuple(req_struct, sizes, org, regTabPtr, disk);
229  shrink_tuple(req_struct, sizes+2, regTabPtr, disk);
230  }
231  else
232  {
233  memcpy(dst, org, 4*regTabPtr->m_offsets[MM].m_fix_header_size);
234  }
235  dst->m_header_bits |= Tuple_header::COPY_TUPLE;
236 
242  assert(sizeof(Local_key) == 8);
243  memcpy(copytuple+0, rowid, sizeof(Local_key));
244 
245  Local_key nil;
246  nil.setNull();
247  memcpy(copytuple+2, &nil, sizeof(nil));
248 
252  if (regFragPtr->m_lcp_keep_list_tail.isNull())
253  {
254  jam();
255  regFragPtr->m_lcp_keep_list_head = opPtrP->m_copy_tuple_location;
256  }
257  else
258  {
259  jam();
260  Uint32 * tail = get_copy_tuple_raw(&regFragPtr->m_lcp_keep_list_tail);
261  Local_key nextptr;
262  memcpy(&nextptr, tail+2, sizeof(Local_key));
263  ndbassert(nextptr.isNull());
264  nextptr = opPtrP->m_copy_tuple_location;
265  memcpy(tail+2, &nextptr, sizeof(Local_key));
266  }
267  regFragPtr->m_lcp_keep_list_tail = opPtrP->m_copy_tuple_location;
268 
272  opPtrP->m_copy_tuple_location.setNull();
273 }
274 
275 #if 0
276 static void dump_buf_hex(unsigned char *p, Uint32 bytes)
277 {
278  char buf[3001];
279  char *q= buf;
280  buf[0]= '\0';
281 
282  for(Uint32 i=0; i<bytes; i++)
283  {
284  if(i==((sizeof(buf)/3)-1))
285  {
286  sprintf(q, "...");
287  break;
288  }
289  sprintf(q+3*i, " %02X", p[i]);
290  }
291  ndbout_c("%8p: %s", p, buf);
292 }
293 #endif
294 
295 void
296 Dbtup::commit_operation(Signal* signal,
297  Uint32 gci_hi,
298  Uint32 gci_lo,
299  Tuple_header* tuple_ptr,
300  PagePtr pagePtr,
301  Operationrec* regOperPtr,
302  Fragrecord* regFragPtr,
303  Tablerec* regTabPtr)
304 {
305  ndbassert(regOperPtr->op_struct.op_type != ZDELETE);
306 
307  Uint32 lcpScan_ptr_i= regFragPtr->m_lcp_scan_op;
308  Uint32 save= tuple_ptr->m_operation_ptr_i;
309  Uint32 bits= tuple_ptr->m_header_bits;
310 
311  Tuple_header *disk_ptr= 0;
312  Tuple_header *copy= get_copy_tuple(&regOperPtr->m_copy_tuple_location);
313 
314  Uint32 copy_bits= copy->m_header_bits;
315 
316  Uint32 fixsize= regTabPtr->m_offsets[MM].m_fix_header_size;
317  Uint32 mm_vars= regTabPtr->m_attributes[MM].m_no_of_varsize;
318  Uint32 mm_dyns= regTabPtr->m_attributes[MM].m_no_of_dynamic;
319  bool update_gci_at_commit = ! regOperPtr->op_struct.m_gci_written;
320  if((mm_vars+mm_dyns) == 0)
321  {
322  jam();
323  memcpy(tuple_ptr, copy, 4*fixsize);
324  disk_ptr= (Tuple_header*)(((Uint32*)copy)+fixsize);
325  }
326  else
327  {
328  jam();
334  Local_key tmp;
335  Var_part_ref *ref= tuple_ptr->get_var_part_ref_ptr(regTabPtr);
336  ref->copyout(&tmp);
337 
338  memcpy(tuple_ptr, copy, 4*fixsize);
339  ref->assign(&tmp);
340 
341  PagePtr vpagePtr;
342  if (copy_bits & Tuple_header::VAR_PART)
343  {
344  jam();
345  ndbassert(bits & Tuple_header::VAR_PART);
346  ndbassert(tmp.m_page_no != RNIL);
347  ndbassert(copy_bits & Tuple_header::COPY_TUPLE);
348 
349  Uint32 *dst= get_ptr(&vpagePtr, *ref);
350  Var_page* vpagePtrP = (Var_page*)vpagePtr.p;
351  Varpart_copy*vp =(Varpart_copy*)copy->get_end_of_fix_part_ptr(regTabPtr);
352  /* The first word of shrunken tuple holds the lenght in words. */
353  Uint32 len = vp->m_len;
354  memcpy(dst, vp->m_data, 4*len);
355 
356  if(copy_bits & Tuple_header::MM_SHRINK)
357  {
358  jam();
359  ndbassert(vpagePtrP->get_entry_len(tmp.m_page_idx) >= len);
360  if (len)
361  {
362  jam();
363  vpagePtrP->shrink_entry(tmp.m_page_idx, len);
364  update_free_page_list(regFragPtr, vpagePtr);
365  }
366  else
367  {
368  jam();
369  free_var_part(regFragPtr, vpagePtr, tmp.m_page_idx);
370  tmp.m_page_no = RNIL;
371  ref->assign(&tmp);
372  copy_bits &= ~(Uint32)Tuple_header::VAR_PART;
373  }
374  }
375  else
376  {
377  jam();
378  ndbassert(vpagePtrP->get_entry_len(tmp.m_page_idx) == len);
379  }
380 
385  disk_ptr = (Tuple_header*)(vp->m_data + len);
386  }
387  else
388  {
389  jam();
390  ndbassert(tmp.m_page_no == RNIL);
391  disk_ptr = (Tuple_header*)copy->get_end_of_fix_part_ptr(regTabPtr);
392  }
393  }
394 
395  if (regTabPtr->m_no_of_disk_attributes &&
396  (copy_bits & Tuple_header::DISK_INLINE))
397  {
398  jam();
399  Local_key key;
400  memcpy(&key, copy->get_disk_ref_ptr(regTabPtr), sizeof(Local_key));
401  Uint32 logfile_group_id= regFragPtr->m_logfile_group_id;
402 
403  PagePtr diskPagePtr = { (Tup_page*)m_pgman_ptr.p, m_pgman_ptr.i };
404  ndbassert(diskPagePtr.p->m_page_no == key.m_page_no);
405  ndbassert(diskPagePtr.p->m_file_no == key.m_file_no);
406  Uint32 sz, *dst;
407  if(copy_bits & Tuple_header::DISK_ALLOC)
408  {
409  jam();
410  disk_page_alloc(signal, regTabPtr, regFragPtr, &key, diskPagePtr, gci_hi);
411  }
412 
413  if(regTabPtr->m_attributes[DD].m_no_of_varsize == 0)
414  {
415  jam();
416  sz= regTabPtr->m_offsets[DD].m_fix_header_size;
417  dst= ((Fix_page*)diskPagePtr.p)->get_ptr(key.m_page_idx, sz);
418  }
419  else
420  {
421  jam();
422  dst= ((Var_page*)diskPagePtr.p)->get_ptr(key.m_page_idx);
423  sz= ((Var_page*)diskPagePtr.p)->get_entry_len(key.m_page_idx);
424  }
425 
426  if(! (copy_bits & Tuple_header::DISK_ALLOC))
427  {
428  jam();
429  disk_page_undo_update(diskPagePtr.p,
430  &key, dst, sz, gci_hi, logfile_group_id);
431  }
432 
433  memcpy(dst, disk_ptr, 4*sz);
434  memcpy(tuple_ptr->get_disk_ref_ptr(regTabPtr), &key, sizeof(Local_key));
435 
436  ndbassert(! (disk_ptr->m_header_bits & Tuple_header::FREE));
437  copy_bits |= Tuple_header::DISK_PART;
438  }
439 
440  if(lcpScan_ptr_i != RNIL && (bits & Tuple_header::ALLOC))
441  {
442  jam();
443  ScanOpPtr scanOp;
444  c_scanOpPool.getPtr(scanOp, lcpScan_ptr_i);
445  Local_key rowid = regOperPtr->m_tuple_location;
446  rowid.m_page_no = pagePtr.p->frag_page_id;
447  if (!is_rowid_lcp_scanned(rowid, *scanOp.p))
448  {
449  jam();
450  copy_bits |= Tuple_header::LCP_SKIP;
451  }
452  }
453 
454  Uint32 clear=
455  Tuple_header::ALLOC | Tuple_header::FREE | Tuple_header::COPY_TUPLE |
456  Tuple_header::DISK_ALLOC | Tuple_header::DISK_INLINE |
457  Tuple_header::MM_SHRINK | Tuple_header::MM_GROWN;
458  copy_bits &= ~(Uint32)clear;
459 
460  tuple_ptr->m_header_bits= copy_bits;
461  tuple_ptr->m_operation_ptr_i= save;
462 
463  if (regTabPtr->m_bits & Tablerec::TR_RowGCI &&
464  update_gci_at_commit)
465  {
466  jam();
467  * tuple_ptr->get_mm_gci(regTabPtr) = gci_hi;
468  if (regTabPtr->m_bits & Tablerec::TR_ExtraRowGCIBits)
469  {
470  Uint32 attrId = regTabPtr->getExtraAttrId<Tablerec::TR_ExtraRowGCIBits>();
471  store_extra_row_bits(attrId, regTabPtr, tuple_ptr, gci_lo,
472  /* truncate */true);
473  }
474  }
475 
476  if (regTabPtr->m_bits & Tablerec::TR_Checksum) {
477  jam();
478  setChecksum(tuple_ptr, regTabPtr);
479  }
480 }
481 
482 void
483 Dbtup::disk_page_commit_callback(Signal* signal,
484  Uint32 opPtrI, Uint32 page_id)
485 {
486  Uint32 hash_value;
487  Uint32 gci_hi, gci_lo;
488  Uint32 transId1, transId2;
489  OperationrecPtr regOperPtr;
490 
491  jamEntry();
492 
493  c_operation_pool.getPtr(regOperPtr, opPtrI);
494  c_lqh->get_op_info(regOperPtr.p->userpointer, &hash_value, &gci_hi, &gci_lo,
495  &transId1, &transId2);
496 
497  TupCommitReq * const tupCommitReq= (TupCommitReq *)signal->getDataPtr();
498 
499  tupCommitReq->opPtr= opPtrI;
500  tupCommitReq->hashValue= hash_value;
501  tupCommitReq->gci_hi= gci_hi;
502  tupCommitReq->gci_lo= gci_lo;
503  tupCommitReq->diskpage = page_id;
504  tupCommitReq->transId1 = transId1;
505  tupCommitReq->transId2 = transId2;
506 
507  regOperPtr.p->op_struct.m_load_diskpage_on_commit= 0;
508  regOperPtr.p->m_commit_disk_callback_page= page_id;
509  m_global_page_pool.getPtr(m_pgman_ptr, page_id);
510 
511  {
512  PagePtr tmp;
513  tmp.i = m_pgman_ptr.i;
514  tmp.p = reinterpret_cast<Page*>(m_pgman_ptr.p);
515  disk_page_set_dirty(tmp);
516  }
517 
518  execTUP_COMMITREQ(signal);
519  if(signal->theData[0] == 0)
520  {
521  jam();
522  c_lqh->tupcommit_conf_callback(signal, regOperPtr.p->userpointer);
523  }
524 }
525 
526 void
527 Dbtup::disk_page_log_buffer_callback(Signal* signal,
528  Uint32 opPtrI,
529  Uint32 unused)
530 {
531  Uint32 hash_value;
532  Uint32 gci_hi, gci_lo;
533  Uint32 transId1, transId2;
534  OperationrecPtr regOperPtr;
535 
536  jamEntry();
537 
538  c_operation_pool.getPtr(regOperPtr, opPtrI);
539  c_lqh->get_op_info(regOperPtr.p->userpointer, &hash_value, &gci_hi, &gci_lo,
540  &transId1, &transId2);
541  Uint32 page= regOperPtr.p->m_commit_disk_callback_page;
542 
543  TupCommitReq * const tupCommitReq= (TupCommitReq *)signal->getDataPtr();
544 
545  tupCommitReq->opPtr= opPtrI;
546  tupCommitReq->hashValue= hash_value;
547  tupCommitReq->gci_hi= gci_hi;
548  tupCommitReq->gci_lo= gci_lo;
549  tupCommitReq->diskpage = page;
550  tupCommitReq->transId1 = transId1;
551  tupCommitReq->transId2 = transId2;
552 
553  ndbassert(regOperPtr.p->op_struct.m_load_diskpage_on_commit == 0);
554  regOperPtr.p->op_struct.m_wait_log_buffer= 0;
555  m_global_page_pool.getPtr(m_pgman_ptr, page);
556 
557  execTUP_COMMITREQ(signal);
558  ndbassert(signal->theData[0] == 0);
559 
560  c_lqh->tupcommit_conf_callback(signal, regOperPtr.p->userpointer);
561 }
562 
563 int Dbtup::retrieve_data_page(Signal *signal,
564  Page_cache_client::Request req,
565  OperationrecPtr regOperPtr)
566 {
567  req.m_callback.m_callbackData= regOperPtr.i;
568  req.m_callback.m_callbackFunction =
569  safe_cast(&Dbtup::disk_page_commit_callback);
570 
571  /*
572  * Consider commit to be correlated. Otherwise pk op + commit makes
573  * the page hot. XXX move to TUP which knows better.
574  */
575  int flags= regOperPtr.p->op_struct.op_type |
576  Page_cache_client::COMMIT_REQ | Page_cache_client::CORR_REQ;
577  Page_cache_client pgman(this, c_pgman);
578  int res= pgman.get_page(signal, req, flags);
579  m_pgman_ptr = pgman.m_ptr;
580 
581  switch(res){
582  case 0:
586  jam();
587  signal->theData[0] = 1;
588  return res;
589  case -1:
590  ndbrequire("NOT YET IMPLEMENTED" == 0);
591  break;
592  default:
593  jam();
594  }
595  {
596  PagePtr tmpptr;
597  tmpptr.i = m_pgman_ptr.i;
598  tmpptr.p = reinterpret_cast<Page*>(m_pgman_ptr.p);
599 
600  disk_page_set_dirty(tmpptr);
601  }
602  regOperPtr.p->m_commit_disk_callback_page= res;
603  regOperPtr.p->op_struct.m_load_diskpage_on_commit= 0;
604 
605  return res;
606 }
607 
608 int Dbtup::retrieve_log_page(Signal *signal,
609  FragrecordPtr regFragPtr,
610  OperationrecPtr regOperPtr)
611 {
612  jam();
618  CallbackPtr cb;
619  cb.m_callbackData= regOperPtr.i;
620  cb.m_callbackIndex = DISK_PAGE_LOG_BUFFER_CALLBACK;
621  Uint32 sz= regOperPtr.p->m_undo_buffer_space;
622 
623  D("Logfile_client - execTUP_COMMITREQ");
624  Logfile_client lgman(this, c_lgman, regFragPtr.p->m_logfile_group_id);
625  int res= lgman.get_log_buffer(signal, sz, &cb);
626  jamEntry();
627  switch(res){
628  case 0:
629  jam();
630  signal->theData[0] = 1;
631  return res;
632  case -1:
633  ndbrequire("NOT YET IMPLEMENTED" == 0);
634  break;
635  default:
636  jam();
637  }
638  regOperPtr.p->op_struct.m_wait_log_buffer= 0;
639 
640  return res;
641 }
642 
646 void
647 Dbtup::findFirstOp(OperationrecPtr & firstPtr)
648 {
649  jam();
650  printf("Detect out-of-order commit(%u) -> ", firstPtr.i);
651  ndbassert(!firstPtr.p->is_first_operation());
652  while(firstPtr.p->prevActiveOp != RNIL)
653  {
654  firstPtr.i = firstPtr.p->prevActiveOp;
655  c_operation_pool.getPtr(firstPtr);
656  }
657  ndbout_c("%u", firstPtr.i);
658 }
659 
660 /* ----------------------------------------------------------------- */
661 /* --------------- COMMIT THIS PART OF A TRANSACTION --------------- */
662 /* ----------------------------------------------------------------- */
663 void Dbtup::execTUP_COMMITREQ(Signal* signal)
664 {
665  FragrecordPtr regFragPtr;
666  OperationrecPtr regOperPtr;
667  TablerecPtr regTabPtr;
668  KeyReqStruct req_struct(this, KRS_COMMIT);
669  TransState trans_state;
670  Uint32 no_of_fragrec, no_of_tablerec;
671 
672  TupCommitReq * const tupCommitReq= (TupCommitReq *)signal->getDataPtr();
673 
674  regOperPtr.i= tupCommitReq->opPtr;
675  Uint32 hash_value= tupCommitReq->hashValue;
676  Uint32 gci_hi = tupCommitReq->gci_hi;
677  Uint32 gci_lo = tupCommitReq->gci_lo;
678  Uint32 transId1 = tupCommitReq->transId1;
679  Uint32 transId2 = tupCommitReq->transId2;
680 
681  jamEntry();
682 
683  c_operation_pool.getPtr(regOperPtr);
684 
685  regFragPtr.i= regOperPtr.p->fragmentPtr;
686  trans_state= get_trans_state(regOperPtr.p);
687 
688  no_of_fragrec= cnoOfFragrec;
689 
690  ndbrequire(trans_state == TRANS_STARTED);
691  ptrCheckGuard(regFragPtr, no_of_fragrec, fragrecord);
692 
693  no_of_tablerec= cnoOfTablerec;
694  regTabPtr.i= regFragPtr.p->fragTableId;
695 
696  req_struct.signal= signal;
697  req_struct.hash_value= hash_value;
698  req_struct.gci_hi = gci_hi;
699  req_struct.gci_lo = gci_lo;
700  /* Put transid in req_struct, so detached triggers can access it */
701  req_struct.trans_id1 = transId1;
702  req_struct.trans_id2 = transId2;
703  regOperPtr.p->m_commit_disk_callback_page = tupCommitReq->diskpage;
704 
705 #ifdef VM_TRACE
706  if (tupCommitReq->diskpage == RNIL)
707  {
708  m_pgman_ptr.i = RNIL;
709  m_pgman_ptr.p = 0;
710  req_struct.m_disk_page_ptr.i = RNIL;
711  req_struct.m_disk_page_ptr.p = 0;
712  }
713 #endif
714 
715  ptrCheckGuard(regTabPtr, no_of_tablerec, tablerec);
716 
717  PagePtr page;
718  Tuple_header* tuple_ptr= (Tuple_header*)
719  get_ptr(&page, &regOperPtr.p->m_tuple_location, regTabPtr.p);
720 
727  if (!regTabPtr.p->tuxCustomTriggers.isEmpty())
728  {
729  if(get_tuple_state(regOperPtr.p) == TUPLE_PREPARED)
730  {
731  jam();
732 
733  OperationrecPtr loopPtr = regOperPtr;
734  if (unlikely(!regOperPtr.p->is_first_operation()))
735  {
736  findFirstOp(loopPtr);
737  }
738 
743  jam();
744  goto first;
745  while(loopPtr.i != RNIL)
746  {
747  c_operation_pool.getPtr(loopPtr);
748  first:
749  executeTuxCommitTriggers(signal,
750  loopPtr.p,
751  regFragPtr.p,
752  regTabPtr.p);
753  set_tuple_state(loopPtr.p, TUPLE_TO_BE_COMMITTED);
754  loopPtr.i = loopPtr.p->nextActiveOp;
755  }
756  }
757  }
758 
759  bool get_page = false;
760  if(regOperPtr.p->op_struct.m_load_diskpage_on_commit)
761  {
762  jam();
763  Page_cache_client::Request req;
764 
769  ndbassert(tuple_ptr->m_operation_ptr_i == regOperPtr.i);
770 
774  if(!regOperPtr.p->m_copy_tuple_location.isNull())
775  {
776  jam();
777  Tuple_header* tmp= get_copy_tuple(&regOperPtr.p->m_copy_tuple_location);
778 
779  memcpy(&req.m_page,
780  tmp->get_disk_ref_ptr(regTabPtr.p), sizeof(Local_key));
781 
782  if (unlikely(regOperPtr.p->op_struct.op_type == ZDELETE &&
783  tmp->m_header_bits & Tuple_header::DISK_ALLOC))
784  {
785  jam();
789  regOperPtr.p->op_struct.m_load_diskpage_on_commit = 0;
790  regOperPtr.p->op_struct.m_wait_log_buffer = 0;
791  disk_page_abort_prealloc(signal, regFragPtr.p,
792  &req.m_page, req.m_page.m_page_idx);
793 
794  D("Logfile_client - execTUP_COMMITREQ");
795  Logfile_client lgman(this, c_lgman, regFragPtr.p->m_logfile_group_id);
796  lgman.free_log_space(regOperPtr.p->m_undo_buffer_space);
797  goto skip_disk;
798  if (0) ndbout_c("insert+delete");
799  jamEntry();
800  goto skip_disk;
801  }
802  }
803  else
804  {
805  jam();
806  // initial delete
807  ndbassert(regOperPtr.p->op_struct.op_type == ZDELETE);
808  memcpy(&req.m_page,
809  tuple_ptr->get_disk_ref_ptr(regTabPtr.p), sizeof(Local_key));
810 
811  ndbassert(tuple_ptr->m_header_bits & Tuple_header::DISK_PART);
812  }
813 
814  if (retrieve_data_page(signal, req, regOperPtr) == 0)
815  {
816  return; // Data page has not been retrieved yet.
817  }
818  get_page = true;
819  }
820 
821  if(regOperPtr.p->op_struct.m_wait_log_buffer)
822  {
823  jam();
828  ndbassert(tuple_ptr->m_operation_ptr_i == regOperPtr.i);
829 
830  if (retrieve_log_page(signal, regFragPtr, regOperPtr) == 0)
831  {
832  return; // Log page has not been retrieved yet.
833  }
834  }
835 
836  assert(tuple_ptr);
837 skip_disk:
838  req_struct.m_tuple_ptr = tuple_ptr;
839 
840  Uint32 nextOp = regOperPtr.p->nextActiveOp;
841  Uint32 prevOp = regOperPtr.p->prevActiveOp;
850  regOperPtr.p->nextActiveOp = RNIL;
851  regOperPtr.p->prevActiveOp = RNIL;
852  if(tuple_ptr->m_operation_ptr_i == regOperPtr.i)
853  {
854  jam();
858  Uint32 disk = regOperPtr.p->m_commit_disk_callback_page;
859  set_commit_change_mask_info(regTabPtr.p, &req_struct, regOperPtr.p);
860  checkDetachedTriggers(&req_struct, regOperPtr.p, regTabPtr.p,
861  disk != RNIL);
862 
863  tuple_ptr->m_operation_ptr_i = RNIL;
864 
865  if (regOperPtr.p->op_struct.op_type == ZDELETE)
866  {
867  jam();
868  if (get_page)
869  {
870  ndbassert(tuple_ptr->m_header_bits & Tuple_header::DISK_PART);
871  }
872  dealloc_tuple(signal, gci_hi, gci_lo, page.p, tuple_ptr,
873  &req_struct, regOperPtr.p, regFragPtr.p, regTabPtr.p);
874  }
875  else if(regOperPtr.p->op_struct.op_type != ZREFRESH)
876  {
877  jam();
878  commit_operation(signal, gci_hi, gci_lo, tuple_ptr, page,
879  regOperPtr.p, regFragPtr.p, regTabPtr.p);
880  }
881  else
882  {
883  jam();
884  commit_refresh(signal, gci_hi, gci_lo, tuple_ptr, page,
885  &req_struct, regOperPtr.p, regFragPtr.p, regTabPtr.p);
886  }
887  }
888 
889  if (nextOp != RNIL)
890  {
891  c_operation_pool.getPtr(nextOp)->prevActiveOp = prevOp;
892  }
893 
894  if (prevOp != RNIL)
895  {
896  c_operation_pool.getPtr(prevOp)->nextActiveOp = nextOp;
897  }
898 
899  if(!regOperPtr.p->m_copy_tuple_location.isNull())
900  {
901  jam();
902  c_undo_buffer.free_copy_tuple(&regOperPtr.p->m_copy_tuple_location);
903  }
904 
905  initOpConnection(regOperPtr.p);
906  signal->theData[0] = 0;
907 }
908 
909 void
910 Dbtup::set_commit_change_mask_info(const Tablerec* regTabPtr,
911  KeyReqStruct * req_struct,
912  const Operationrec * regOperPtr)
913 {
914  Uint32 masklen = (regTabPtr->m_no_of_attributes + 31) >> 5;
915  if (regOperPtr->m_copy_tuple_location.isNull())
916  {
917  ndbassert(regOperPtr->op_struct.op_type == ZDELETE);
918  req_struct->changeMask.set();
919  }
920  else
921  {
922  Uint32 * dst = req_struct->changeMask.rep.data;
923  Uint32 * rawptr = get_copy_tuple_raw(&regOperPtr->m_copy_tuple_location);
924  ChangeMask * maskptr = get_change_mask_ptr(rawptr);
925  Uint32 cols = maskptr->m_cols;
926  if (cols == regTabPtr->m_no_of_attributes)
927  {
928  memcpy(dst, maskptr->m_mask, 4*masklen);
929  }
930  else
931  {
932  ndbassert(regTabPtr->m_no_of_attributes > cols); // no drop column
933  memcpy(dst, maskptr->m_mask, 4*((cols + 31) >> 5));
934  req_struct->changeMask.setRange(cols,
935  regTabPtr->m_no_of_attributes - cols);
936  }
937  }
938 }
939 
940 void
941 Dbtup::commit_refresh(Signal* signal,
942  Uint32 gci_hi,
943  Uint32 gci_lo,
944  Tuple_header* tuple_ptr,
945  PagePtr pagePtr,
946  KeyReqStruct * req_struct,
947  Operationrec* regOperPtr,
948  Fragrecord* regFragPtr,
949  Tablerec* regTabPtr)
950 {
951  /* Committing a refresh operation.
952  * Refresh of an existing row looks like an update
953  * and can commit normally.
954  * Refresh of a non-existing row looks like an Insert which
955  * is 'undone' at commit time.
956  * This is achieved by making special calls to ACC to get
957  * it to forget, before deallocating the tuple locally.
958  */
959  switch(regOperPtr->m_copy_tuple_location.m_file_no){
960  case Operationrec::RF_SINGLE_NOT_EXIST:
961  case Operationrec::RF_MULTI_NOT_EXIST:
962  break;
963  case Operationrec::RF_SINGLE_EXIST:
964  case Operationrec::RF_MULTI_EXIST:
965  // "Normal" update
966  commit_operation(signal, gci_hi, gci_lo, tuple_ptr, pagePtr,
967  regOperPtr, regFragPtr, regTabPtr);
968  return;
969 
970  default:
971  ndbrequire(false);
972  }
973 
974  Local_key key = regOperPtr->m_tuple_location;
975  key.m_page_no = pagePtr.p->frag_page_id;
976 
980  c_lqh->accremoverow(signal, regOperPtr->userpointer, &key);
981  dealloc_tuple(signal, gci_hi, gci_lo, pagePtr.p, tuple_ptr,
982  req_struct, regOperPtr, regFragPtr, regTabPtr);
983 }