Ndbif.cpp 42.3 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
/* Copyright (C) 2003 MySQL AB

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 2 of the License, or
   (at your option) any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */


18 19
#include <ndb_global.h>

20 21 22 23
#include "NdbApiSignal.hpp"
#include "NdbImpl.hpp"
#include "NdbOperation.hpp"
#include "NdbIndexOperation.hpp"
unknown's avatar
unknown committed
24
#include "NdbScanOperation.hpp"
25 26 27 28 29 30 31 32 33 34 35 36
#include "NdbConnection.hpp"
#include "NdbRecAttr.hpp"
#include "NdbReceiver.hpp"
#include "API.hpp"

#include <signaldata/TcCommit.hpp>
#include <signaldata/TcKeyFailConf.hpp>
#include <signaldata/TcKeyConf.hpp>
#include <signaldata/TestOrd.hpp>
#include <signaldata/CreateIndx.hpp>
#include <signaldata/DropIndx.hpp>
#include <signaldata/TcIndx.hpp>
unknown's avatar
unknown committed
37 38 39
#include <signaldata/TransIdAI.hpp>
#include <signaldata/ScanFrag.hpp>
#include <signaldata/ScanTab.hpp>
40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57

#include <ndb_limits.h>
#include <NdbOut.hpp>
#include <NdbTick.h>


/******************************************************************************
 * int init( int aNrOfCon, int aNrOfOp );
 *
 * Return Value:   Return 0 : init was successful.
 *                Return -1: In all other case.  
 * Parameters:	aNrOfCon : Number of connections offered to the application.
 *		aNrOfOp : Number of operations offered to the application.
 * Remark:		Create pointers and idle list Synchronous.
 ****************************************************************************/ 
int
Ndb::init(int aMaxNoOfTransactions)
{
58 59
  DBUG_ENTER("Ndb::init");

60 61 62 63 64 65 66 67 68 69 70 71 72 73
  int i;
  int aNrOfCon;
  int aNrOfOp;
  int tMaxNoOfTransactions;
  NdbApiSignal* tSignal[16];	// Initiate free list of 16 signal objects
  if (theInitState != NotInitialised) {
    switch(theInitState){
    case InitConfigError:
      theError.code = 4117;
      break;
    default:
      theError.code = 4104;
      break;
    }
74
    DBUG_RETURN(-1);
75 76 77 78 79 80 81
  }//if
  theInitState = StartingInit;
  TransporterFacade * theFacade =  TransporterFacade::instance();
  theFacade->lock_mutex();
  
  const int tBlockNo = theFacade->open(this,
                                       executeMessage, 
82
                                       statusMessage);  
83 84 85
  if ( tBlockNo == -1 ) {
    theError.code = 4105;
    theFacade->unlock_mutex();
86
    DBUG_RETURN(-1); // no more free blocknumbers
87 88 89
  }//if
  
  theNdbBlockNumber = tBlockNo;
unknown's avatar
unknown committed
90

91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140
  theFacade->unlock_mutex();
  
  theDictionary->setTransporter(this, theFacade);
  
  aNrOfCon = theNoOfDBnodes;
  aNrOfOp = 2*theNoOfDBnodes;
  
  // Create connection object in a linked list 
  if((createConIdleList(aNrOfCon)) == -1){
    theError.code = 4000;
    goto error_handler;
  }
  
  // Create operations in a linked list
  if((createOpIdleList(aNrOfOp)) == -1){       
    theError.code = 4000;
    goto error_handler;
  }
  
  tMaxNoOfTransactions = aMaxNoOfTransactions * 3;
  if (tMaxNoOfTransactions > 1024) {
    tMaxNoOfTransactions = 1024;
  }//if
  theMaxNoOfTransactions = tMaxNoOfTransactions;
  
  thePreparedTransactionsArray = new NdbConnection* [tMaxNoOfTransactions];
  theSentTransactionsArray = new NdbConnection* [tMaxNoOfTransactions];
  theCompletedTransactionsArray = new NdbConnection* [tMaxNoOfTransactions];
  
  if ((thePreparedTransactionsArray == NULL) ||
      (theSentTransactionsArray == NULL) ||
      (theCompletedTransactionsArray == NULL)) {
    goto error_handler;
  }//if
  
  for (i = 0; i < tMaxNoOfTransactions; i++) {
    thePreparedTransactionsArray[i] = NULL;
    theSentTransactionsArray[i] = NULL;
    theCompletedTransactionsArray[i] = NULL;
  }//for     
  for (i = 0; i < 16; i++){
    tSignal[i] = getSignal();
    if(tSignal[i] == NULL) {
      theError.code = 4000;
      goto error_handler;
    }
  }
  for (i = 0; i < 16; i++)
    releaseSignal(tSignal[i]);
  theInitState = Initialised; 
141
  DBUG_RETURN(0);
142 143 144 145 146 147 148 149 150 151 152 153 154 155
  
error_handler:
  ndbout << "error_handler" << endl;
  releaseTransactionArrays();
  while ( theConIdleList != NULL )
    freeNdbCon();
  while ( theSignalIdleList != NULL )
    freeSignal();
  while (theRecAttrIdleList != NULL)
    freeRecAttr(); 
  while (theOpIdleList != NULL)
    freeOperation();
  
  delete theDictionary;
156
  TransporterFacade::instance()->close(theNdbBlockNumber, 0);
157
  DBUG_RETURN(-1);
158 159 160 161 162
}

void
Ndb::releaseTransactionArrays()
{
163
  DBUG_ENTER("Ndb::releaseTransactionArrays");
164 165 166 167 168 169 170 171 172
  if (thePreparedTransactionsArray != NULL) {
    delete [] thePreparedTransactionsArray;
  }//if
  if (theSentTransactionsArray != NULL) {
    delete [] theSentTransactionsArray;
  }//if
  if (theCompletedTransactionsArray != NULL) {
    delete [] theCompletedTransactionsArray;
  }//if
173
  DBUG_VOID_RETURN;
174 175 176 177 178 179 180 181 182 183 184
}//Ndb::releaseTransactionArrays()

void
Ndb::executeMessage(void* NdbObject,
                    NdbApiSignal * aSignal,
                    LinearSectionPtr ptr[3])
{
  Ndb* tNdb = (Ndb*)NdbObject;
  tNdb->handleReceivedSignal(aSignal, ptr);
}

185 186 187
void Ndb::connected(Uint32 ref)
{
  theMyRef= ref;
188
  Uint32 tmpTheNode= refToNode(ref);
unknown's avatar
unknown committed
189 190
  Uint64 tBlockNo= refToBlock(ref);
  if (theNdbBlockNumber >= 0){
191
    assert(theMyRef == numberToRef(theNdbBlockNumber, tmpTheNode));
unknown's avatar
unknown committed
192 193
  }
  
194 195 196 197 198 199 200 201 202
  TransporterFacade * theFacade =  TransporterFacade::instance();
  int i;
  theNoOfDBnodes= 0;
  for (i = 1; i < MAX_NDB_NODES; i++){
    if (theFacade->getIsDbNode(i)){
      theDBnodes[theNoOfDBnodes] = i;
      theNoOfDBnodes++;
    }
  }
unknown's avatar
unknown committed
203
  theFirstTransId = ((Uint64)tBlockNo << 52)+
204
    ((Uint64)tmpTheNode << 40);
205 206
  theFirstTransId += theFacade->m_max_trans_id;
  //      assert(0);
unknown's avatar
unknown committed
207
  DBUG_PRINT("info",("connected with ref=%x, id=%d, no_db_nodes=%d, first_trans_id=%lx",
208
		     theMyRef,
209
		     tmpTheNode,
210 211 212 213 214 215
		     theNoOfDBnodes,
		     theFirstTransId));
  startTransactionNodeSelectionData.init(theNoOfDBnodes, theDBnodes);
  theCommitAckSignal = new NdbApiSignal(theMyRef);

  theDictionary->m_receiver.m_reference= theMyRef;
216
  theNode= tmpTheNode; // flag that Ndb object is initialized
217 218
}

219
void
220
Ndb::statusMessage(void* NdbObject, Uint32 a_node, bool alive, bool nfComplete)
221
{
222
  DBUG_ENTER("Ndb::statusMessage");
223 224 225
  Ndb* tNdb = (Ndb*)NdbObject;
  if (alive) {
    if (nfComplete) {
226 227
      tNdb->connected(a_node);
      DBUG_VOID_RETURN;
228 229 230 231 232 233 234 235 236 237
    }//if
  } else {
    if (nfComplete) {
      tNdb->report_node_failure_completed(a_node);
    } else {
      tNdb->report_node_failure(a_node);
    }//if
  }//if
  NdbDictInterface::execNodeStatus(&tNdb->theDictionary->m_receiver,
				   a_node, alive, nfComplete);
238
  DBUG_VOID_RETURN;
239 240 241 242 243 244 245 246 247 248 249 250 251
}

void
Ndb::report_node_failure(Uint32 node_id)
{
  /**
   * We can only set the state here since this object can execute 
   * simultaneously. 
   * 
   * This method is only called by ClusterMgr (via lots of methods)
   */
  the_release_ind[node_id] = 1;
  theWaiter.nodeFail(node_id);
252
  return;
253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274
}//Ndb::report_node_failure()


void
Ndb::report_node_failure_completed(Uint32 node_id)
{
  abortTransactionsAfterNodeFailure(node_id);

}//Ndb::report_node_failure_completed()

/***************************************************************************
void abortTransactionsAfterNodeFailure();

Remark:   Abort all transactions in theSentTransactionsArray after connection 
          to one node has failed
****************************************************************************/
void	
Ndb::abortTransactionsAfterNodeFailure(Uint16 aNodeId)
{  
  Uint32 tNoSentTransactions = theNoOfSentTransactions;
  for (int i = tNoSentTransactions - 1; i >= 0; i--) {
    NdbConnection* localCon = theSentTransactionsArray[i];
275
    if (localCon->getConnectedNodeId() == aNodeId) {
unknown's avatar
unknown committed
276
      const NdbConnection::SendStatusType sendStatus = localCon->theSendStatus;
277 278
      if (sendStatus == NdbConnection::sendTC_OP || 
	  sendStatus == NdbConnection::sendTC_COMMIT) {
279 280 281 282 283 284 285
        /*
        A transaction was interrupted in the prepare phase by a node
        failure. Since the transaction was not found in the phase
        after the node failure it cannot have been committed and
        we report a normal node failure abort.
        */
	localCon->setOperationErrorCodeAbort(4010);
unknown's avatar
unknown committed
286 287
        localCon->theCompletionStatus = NdbConnection::CompletedFailure;
      } else if (sendStatus == NdbConnection::sendTC_ROLLBACK) {
288 289 290 291
        /*
        We aimed for abort and abort we got even if it was by a node
        failure. We will thus report it as a success.
        */
unknown's avatar
unknown committed
292
        localCon->theCompletionStatus = NdbConnection::CompletedSuccess;
293 294 295 296 297
      } else {
#ifdef VM_TRACE
        printState("abortTransactionsAfterNodeFailure %x", this);
        abort();
#endif
298
      }
299 300 301 302 303
      /*
      All transactions arriving here have no connection to the kernel
      intact since the node was failing and they were aborted. Thus we
      set commit state to Aborted and set state to release on close.
      */
unknown's avatar
unknown committed
304
      localCon->theCommitStatus = NdbConnection::Aborted;
305 306
      localCon->theReleaseOnClose = true;
      completedTransaction(localCon);
307 308 309 310 311
    } 
    else if(localCon->report_node_failure(aNodeId));
    {
      completedTransaction(localCon);
    }
312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327
  }//for
  return;
}//Ndb::abortTransactionsAfterNodeFailure()

/****************************************************************************
void handleReceivedSignal(NdbApiSignal* aSignal);

Parameters:     aSignal: The signal object.
Remark:         Send all operations belonging to this connection. 
*****************************************************************************/
void	
Ndb::handleReceivedSignal(NdbApiSignal* aSignal, LinearSectionPtr ptr[3])
{
  NdbOperation* tOp;
  NdbIndexOperation* tIndexOp;
  NdbConnection* tCon;
unknown's avatar
unknown committed
328
  int tReturnCode = -1;
329 330 331 332
  const Uint32* tDataPtr = aSignal->getDataPtr();
  const Uint32 tWaitState = theWaiter.m_state;
  const Uint32 tSignalNumber = aSignal->readSignalNumber();
  const Uint32 tFirstData = *tDataPtr;
unknown's avatar
unknown committed
333 334
  const Uint32 tLen = aSignal->getLength();
  void * tFirstDataPtr;
335 336

  /*
unknown's avatar
unknown committed
337 338 339 340 341 342 343
    In order to support 64 bit processes in the application we need to use
    id's rather than a direct pointer to the object used. It is also a good
    idea that one cannot corrupt the application code by sending a corrupt
    memory pointer.
    
    All signals received by the API requires the first data word to be such
    an id to the receiving object.
344
  */
unknown's avatar
unknown committed
345
  
346 347 348
  switch (tSignalNumber){
  case GSN_TCKEYCONF:
    {
unknown's avatar
unknown committed
349
      tFirstDataPtr = int2void(tFirstData);
350 351 352 353 354 355 356
      if (tFirstDataPtr == 0) goto InvalidSignal;

      const TcKeyConf * const keyConf = (TcKeyConf *)tDataPtr;
      const BlockReference aTCRef = aSignal->theSendersBlockRef;

      tCon = void2con(tFirstDataPtr);
      if ((tCon->checkMagicNumber() == 0) &&
unknown's avatar
unknown committed
357
          (tCon->theSendStatus == NdbConnection::sendTC_OP)) {
unknown's avatar
unknown committed
358
        tReturnCode = tCon->receiveTCKEYCONF(keyConf, tLen);
359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375
        if (tReturnCode != -1) {
          completedTransaction(tCon);
        }//if

	if(TcKeyConf::getMarkerFlag(keyConf->confInfo)){
	  NdbConnection::sendTC_COMMIT_ACK(theCommitAckSignal,
					   keyConf->transId1, 
					   keyConf->transId2,
					   aTCRef);
	}
      
	return;
      }//if
      goto InvalidSignal;
      
      return;
    }
376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396
  case GSN_TRANSID_AI:{
    tFirstDataPtr = int2void(tFirstData);
    NdbReceiver* tRec;
    if (tFirstDataPtr && (tRec = void2rec(tFirstDataPtr)) && 
	tRec->checkMagicNumber() && (tCon = tRec->getTransaction()) &&
	tCon->checkState_TransId(((const TransIdAI*)tDataPtr)->transId)){
      Uint32 com;
      if(aSignal->m_noOfSections > 0){
	com = tRec->execTRANSID_AI(ptr[0].p, ptr[0].sz);
      } else {
	com = tRec->execTRANSID_AI(tDataPtr + TransIdAI::HeaderLength, 
				   tLen - TransIdAI::HeaderLength);
      }
      
      if(com == 1){
	switch(tRec->getType()){
	case NdbReceiver::NDB_OPERATION:
	case NdbReceiver::NDB_INDEX_OPERATION:
	  if(tCon->OpCompleteSuccess() != -1){
	    completedTransaction(tCon);
	    return;
397
	  }
398 399 400
	  break;
	case NdbReceiver::NDB_SCANRECEIVER:
	  tCon->theScanningOp->receiver_delivered(tRec);
unknown's avatar
unknown committed
401 402
	  theWaiter.m_state = (((WaitSignalType) tWaitState) == WAIT_SCAN ? 
			       (Uint32) NO_WAIT : tWaitState);
403 404 405
	  break;
	default:
	  goto InvalidSignal;
406 407
	}
      }
408 409 410 411 412 413 414
      break;
    } else {
      /**
       * This is ok as transaction can have been aborted before TRANSID_AI
       * arrives (if TUP on  other node than TC)
       */
      return;
415
    }
416
  }
417 418
  case GSN_TCKEY_FAILCONF:
    {
unknown's avatar
unknown committed
419
      tFirstDataPtr = int2void(tFirstData);
unknown's avatar
unknown committed
420
      const TcKeyFailConf * failConf = (TcKeyFailConf *)tDataPtr;
421
      const BlockReference aTCRef = aSignal->theSendersBlockRef;
unknown's avatar
unknown committed
422 423 424 425 426 427 428 429 430 431 432 433
      if (tFirstDataPtr != 0){
	tOp = void2rec_op(tFirstDataPtr);
	
	if (tOp->checkMagicNumber(false) == 0) {
	  tCon = tOp->theNdbCon;
	  if (tCon != NULL) {
	    if ((tCon->theSendStatus == NdbConnection::sendTC_OP) ||
		(tCon->theSendStatus == NdbConnection::sendTC_COMMIT)) {
	      tReturnCode = tCon->receiveTCKEY_FAILCONF(failConf);
	      if (tReturnCode != -1) {
		completedTransaction(tCon);
	      }//if
434
	    }//if
unknown's avatar
unknown committed
435 436 437 438 439 440 441
	  }
	}
      } else {
#ifdef VM_TRACE
	ndbout_c("Recevied TCKEY_FAILCONF wo/ operation");
#endif
      }
442 443 444 445 446 447 448 449 450 451
      if(tFirstData & 1){
	NdbConnection::sendTC_COMMIT_ACK(theCommitAckSignal,
					 failConf->transId1, 
					 failConf->transId2,
					 aTCRef);
      }
      return;
    }
  case GSN_TCKEY_FAILREF:
    {
unknown's avatar
unknown committed
452
      tFirstDataPtr = int2void(tFirstData);
unknown's avatar
unknown committed
453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473
      if(tFirstDataPtr != 0){
	tOp = void2rec_op(tFirstDataPtr);
	if (tOp->checkMagicNumber() == 0) {
	  tCon = tOp->theNdbCon;
	  if (tCon != NULL) {
	    if ((tCon->theSendStatus == NdbConnection::sendTC_OP) ||
		(tCon->theSendStatus == NdbConnection::sendTC_ROLLBACK)) {
	      tReturnCode = tCon->receiveTCKEY_FAILREF(aSignal);
	      if (tReturnCode != -1) {
		completedTransaction(tCon);
		return;
	      }//if
	    }//if
	  }//if
	}//if
      } else {
#ifdef VM_TRACE
	ndbout_c("Recevied TCKEY_FAILREF wo/ operation");
#endif
      }
      break;
474 475 476
    }
  case GSN_TCKEYREF:
    {
unknown's avatar
unknown committed
477
      tFirstDataPtr = int2void(tFirstData);
478 479 480 481 482 483
      if (tFirstDataPtr == 0) goto InvalidSignal;

      tOp = void2rec_op(tFirstDataPtr);
      if (tOp->checkMagicNumber() == 0) {
	tCon = tOp->theNdbCon;
	if (tCon != NULL) {
unknown's avatar
unknown committed
484
	  if (tCon->theSendStatus == NdbConnection::sendTC_OP) {
485 486 487
	    tReturnCode = tOp->receiveTCKEYREF(aSignal);
	    if (tReturnCode != -1) {
	      completedTransaction(tCon);
unknown's avatar
unknown committed
488
	      return;
489
	    }//if
unknown's avatar
unknown committed
490
	    break;
491 492 493 494 495 496 497 498
	  }//if
	}//if
      } //if
      goto InvalidSignal;
      return;
    } 
  case GSN_TC_COMMITCONF:
    {
unknown's avatar
unknown committed
499
      tFirstDataPtr = int2void(tFirstData);
500 501 502 503 504 505 506
      if (tFirstDataPtr == 0) goto InvalidSignal;

      const TcCommitConf * const commitConf = (TcCommitConf *)tDataPtr;
      const BlockReference aTCRef = aSignal->theSendersBlockRef;
      
      tCon = void2con(tFirstDataPtr);
      if ((tCon->checkMagicNumber() == 0) &&
unknown's avatar
unknown committed
507
	  (tCon->theSendStatus == NdbConnection::sendTC_COMMIT)) {
508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526
	tReturnCode = tCon->receiveTC_COMMITCONF(commitConf);
	if (tReturnCode != -1) {
	  completedTransaction(tCon);
	}//if

	if(tFirstData & 1){
	  NdbConnection::sendTC_COMMIT_ACK(theCommitAckSignal,
					   commitConf->transId1, 
					   commitConf->transId2,
					   aTCRef);
	}
	return;
      }
      goto InvalidSignal;
      return;
    }

  case GSN_TC_COMMITREF:
    {
unknown's avatar
unknown committed
527
      tFirstDataPtr = int2void(tFirstData);
528 529 530 531
      if (tFirstDataPtr == 0) goto InvalidSignal;

      tCon = void2con(tFirstDataPtr);
      if ((tCon->checkMagicNumber() == 0) &&
unknown's avatar
unknown committed
532
	  (tCon->theSendStatus == NdbConnection::sendTC_COMMIT)) {
533 534 535 536 537 538 539 540 541
	tReturnCode = tCon->receiveTC_COMMITREF(aSignal);
	if (tReturnCode != -1) {
	  completedTransaction(tCon);
	}//if
      }//if
      return;
    }
  case GSN_TCROLLBACKCONF:
    {
unknown's avatar
unknown committed
542
      tFirstDataPtr = int2void(tFirstData);
543 544 545 546
      if (tFirstDataPtr == 0) goto InvalidSignal;

      tCon = void2con(tFirstDataPtr);
      if ((tCon->checkMagicNumber() == 0) &&
unknown's avatar
unknown committed
547
	  (tCon->theSendStatus == NdbConnection::sendTC_ROLLBACK)) {
548 549 550 551 552 553 554 555 556
	tReturnCode = tCon->receiveTCROLLBACKCONF(aSignal);
	if (tReturnCode != -1) {
	  completedTransaction(tCon);
	}//if
      }//if
      return;
    }
  case GSN_TCROLLBACKREF:
    {
unknown's avatar
unknown committed
557
      tFirstDataPtr = int2void(tFirstData);
558 559 560 561
      if (tFirstDataPtr == 0) goto InvalidSignal;

      tCon = void2con(tFirstDataPtr);
      if ((tCon->checkMagicNumber() == 0) &&
unknown's avatar
unknown committed
562
	  (tCon->theSendStatus == NdbConnection::sendTC_ROLLBACK)) {
563 564 565 566 567 568 569 570 571
	tReturnCode = tCon->receiveTCROLLBACKREF(aSignal);
	if (tReturnCode != -1) {
	  completedTransaction(tCon);
	}//if
      }//if
      return;
    }
  case GSN_TCROLLBACKREP:
    {
unknown's avatar
unknown committed
572
      tFirstDataPtr = int2void(tFirstData);
573 574 575 576 577 578 579 580 581 582 583 584 585
      if (tFirstDataPtr == 0) goto InvalidSignal;

      tCon = void2con(tFirstDataPtr);
      if (tCon->checkMagicNumber() == 0) {
	tReturnCode = tCon->receiveTCROLLBACKREP(aSignal);
	if (tReturnCode != -1) {
	  completedTransaction(tCon);
	}//if
      }//if
      return;
    }
  case GSN_TCSEIZECONF:
    {
unknown's avatar
unknown committed
586
      tFirstDataPtr = int2void(tFirstData);
587 588 589
      if (tFirstDataPtr == 0) goto InvalidSignal;

      if (tWaitState != WAIT_TC_SEIZE) {
unknown's avatar
unknown committed
590
	goto InvalidSignal;
591 592 593
      }//if
      tCon = void2con(tFirstDataPtr);
      if (tCon->checkMagicNumber() != 0) {
unknown's avatar
unknown committed
594
	goto InvalidSignal;
595 596 597 598 599
      }//if
      tReturnCode = tCon->receiveTCSEIZECONF(aSignal);
      if (tReturnCode != -1) {
	theWaiter.m_state = NO_WAIT;
      } else {
unknown's avatar
unknown committed
600
	goto InvalidSignal;
601 602 603 604 605
      }//if
      break;
    }
  case GSN_TCSEIZEREF:
    {
unknown's avatar
unknown committed
606
      tFirstDataPtr = int2void(tFirstData);
607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625
      if (tFirstDataPtr == 0) goto InvalidSignal;

      if (tWaitState != WAIT_TC_SEIZE) {
	return;
      }//if
      tCon = void2con(tFirstDataPtr);
      if (tCon->checkMagicNumber() != 0) {
	return;
      }//if
      tReturnCode = tCon->receiveTCSEIZEREF(aSignal);
      if (tReturnCode != -1) {
	theWaiter.m_state = NO_WAIT;
      } else {
        return;
      }//if
      break;
    }
  case GSN_TCRELEASECONF:
    {
unknown's avatar
unknown committed
626
      tFirstDataPtr = int2void(tFirstData);
627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643
      if (tFirstDataPtr == 0) goto InvalidSignal;

      if (tWaitState != WAIT_TC_RELEASE) {
	goto InvalidSignal;
      }//if
      tCon = void2con(tFirstDataPtr);
      if (tCon->checkMagicNumber() != 0) {
	goto InvalidSignal;
      }//if
      tReturnCode = tCon->receiveTCRELEASECONF(aSignal);
      if (tReturnCode != -1) {
	theWaiter.m_state = NO_WAIT;
      }//if
      break;
    } 
  case GSN_TCRELEASEREF:
    {
unknown's avatar
unknown committed
644
      tFirstDataPtr = int2void(tFirstData);
645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697
      if (tFirstDataPtr == 0) goto InvalidSignal;

      if (tWaitState != WAIT_TC_RELEASE) {
	goto InvalidSignal;
      }//if
      tCon = void2con(tFirstDataPtr);
      if (tCon->checkMagicNumber() != 0) {
	goto InvalidSignal;
      }//if
      tReturnCode = tCon->receiveTCRELEASEREF(aSignal);
      if (tReturnCode != -1) {
	theWaiter.m_state = NO_WAIT;
      }//if
      break;
    }
      
  case GSN_GET_TABINFOREF:
  case GSN_GET_TABINFO_CONF:
  case GSN_CREATE_TABLE_REF:
  case GSN_CREATE_TABLE_CONF:
  case GSN_DROP_TABLE_CONF:
  case GSN_DROP_TABLE_REF:
  case GSN_ALTER_TABLE_CONF:
  case GSN_ALTER_TABLE_REF:
  case GSN_CREATE_INDX_CONF:
  case GSN_CREATE_INDX_REF:
  case GSN_DROP_INDX_CONF:
  case GSN_DROP_INDX_REF:
  case GSN_CREATE_EVNT_CONF:
  case GSN_CREATE_EVNT_REF:
  case GSN_DROP_EVNT_CONF:
  case GSN_DROP_EVNT_REF:
  case GSN_LIST_TABLES_CONF:
    NdbDictInterface::execSignal(&theDictionary->m_receiver,
				 aSignal, ptr);
    break;
    
  case GSN_SUB_META_DATA:
  case GSN_SUB_REMOVE_CONF:
  case GSN_SUB_REMOVE_REF:
    break; // ignore these signals
  case GSN_SUB_GCP_COMPLETE_REP:
  case GSN_SUB_START_CONF:
  case GSN_SUB_START_REF:
  case GSN_SUB_TABLE_DATA:
  case GSN_SUB_STOP_CONF:
  case GSN_SUB_STOP_REF:
    NdbDictInterface::execSignal(&theDictionary->m_receiver,
				 aSignal, ptr);
    break;

  case GSN_DIHNDBTAMPER:
    {
unknown's avatar
unknown committed
698 699
      tFirstDataPtr = int2void(tFirstData);
      if (tFirstDataPtr == 0) goto InvalidSignal;
unknown's avatar
unknown committed
700
      
unknown's avatar
unknown committed
701 702 703 704 705 706
      if (tWaitState != WAIT_NDB_TAMPER)
	return;
      tCon = void2con(tFirstDataPtr);
      if (tCon->checkMagicNumber() != 0)
	return;
      tReturnCode = tCon->receiveDIHNDBTAMPER(aSignal);
unknown's avatar
unknown committed
707 708 709
      if (tReturnCode != -1)
	theWaiter.m_state = NO_WAIT;
      break;
710
    }
unknown's avatar
unknown committed
711 712 713 714 715 716 717 718 719 720
  case GSN_SCAN_TABCONF:
    {
      tFirstDataPtr = int2void(tFirstData);
      assert(tFirstDataPtr);
      assert(void2con(tFirstDataPtr));
      assert(void2con(tFirstDataPtr)->checkMagicNumber() == 0);
      if(tFirstDataPtr && 
	 (tCon = void2con(tFirstDataPtr)) && (tCon->checkMagicNumber() == 0)){
	
	if(aSignal->m_noOfSections > 0){
721 722
	  tReturnCode = tCon->receiveSCAN_TABCONF(aSignal, 
						  ptr[0].p, ptr[0].sz);
unknown's avatar
unknown committed
723 724 725 726 727 728
	} else {
	  tReturnCode = 
	    tCon->receiveSCAN_TABCONF(aSignal, 
				      tDataPtr + ScanTabConf::SignalLength, 
				      tLen - ScanTabConf::SignalLength);
	}
unknown's avatar
unknown committed
729
	if (tReturnCode != -1 && tWaitState == WAIT_SCAN)
unknown's avatar
unknown committed
730 731 732 733 734 735
	  theWaiter.m_state = NO_WAIT;
	break;
      } else {
	goto InvalidSignal;
      }
    }
736
  case GSN_SCAN_TABREF:
unknown's avatar
unknown committed
737 738 739 740 741 742 743 744 745 746 747
    {
      tFirstDataPtr = int2void(tFirstData);
      if (tFirstDataPtr == 0) goto InvalidSignal;
      
      tCon = void2con(tFirstDataPtr);
      
      assert(tFirstDataPtr != 0 && 
	     void2con(tFirstDataPtr)->checkMagicNumber() == 0);
      
      if (tCon->checkMagicNumber() == 0){
	tReturnCode = tCon->receiveSCAN_TABREF(aSignal);
unknown's avatar
unknown committed
748
	if (tReturnCode != -1 && tWaitState == WAIT_SCAN){
unknown's avatar
unknown committed
749 750 751
	  theWaiter.m_state = NO_WAIT;
	}
	break;
752
      }
unknown's avatar
unknown committed
753
      goto InvalidSignal;
754 755
    }
  case GSN_KEYINFO20: {
unknown's avatar
unknown committed
756
    tFirstDataPtr = int2void(tFirstData);
757 758 759 760 761
    NdbReceiver* tRec;
    if (tFirstDataPtr && (tRec = void2rec(tFirstDataPtr)) &&
	tRec->checkMagicNumber() && (tCon = tRec->getTransaction()) &&
	tCon->checkState_TransId(&((const KeyInfo20*)tDataPtr)->transId1)){
      
unknown's avatar
unknown committed
762 763 764 765 766 767 768 769 770 771 772 773
      Uint32 len = ((const KeyInfo20*)tDataPtr)->keyLen;
      Uint32 info = ((const KeyInfo20*)tDataPtr)->scanInfo_Node;
      int com = -1;
      if(aSignal->m_noOfSections > 0 && len == ptr[0].sz){
	com = tRec->execKEYINFO20(info, ptr[0].p, len);
      } else if(len == tLen - KeyInfo20::HeaderLength){
	com = tRec->execKEYINFO20(info, tDataPtr+KeyInfo20::HeaderLength, len);
      }
      
      switch(com){
      case 1:
	tCon->theScanningOp->receiver_delivered(tRec);
unknown's avatar
unknown committed
774 775
	theWaiter.m_state = (((WaitSignalType) tWaitState) == WAIT_SCAN ? 
			      (Uint32) NO_WAIT : tWaitState);
unknown's avatar
unknown committed
776 777 778 779 780 781 782
	break;
      case 0:
	break;
      case -1:
	goto InvalidSignal;
      }
      break;
783 784 785 786 787 788
    } else {
      /**
       * This is ok as transaction can have been aborted before KEYINFO20
       * arrives (if TUP on  other node than TC)
       */
      return;
unknown's avatar
unknown committed
789
    }
790 791
  }
  case GSN_TCINDXCONF:{
unknown's avatar
unknown committed
792
    tFirstDataPtr = int2void(tFirstData);
793 794 795 796 797 798
    if (tFirstDataPtr == 0) goto InvalidSignal;

    const TcIndxConf * const indxConf = (TcIndxConf *)tDataPtr;
    const BlockReference aTCRef = aSignal->theSendersBlockRef;
    tCon = void2con(tFirstDataPtr);
    if ((tCon->checkMagicNumber() == 0) &&
unknown's avatar
unknown committed
799
	(tCon->theSendStatus == NdbConnection::sendTC_OP)) {
unknown's avatar
unknown committed
800
      tReturnCode = tCon->receiveTCINDXCONF(indxConf, tLen);
801 802 803 804 805 806 807 808 809 810 811
      if (tReturnCode != -1) { 
	completedTransaction(tCon);
      }//if
    }//if
    
    if(TcIndxConf::getMarkerFlag(indxConf->confInfo)){
      NdbConnection::sendTC_COMMIT_ACK(theCommitAckSignal,
				       indxConf->transId1, 
				       indxConf->transId2,
				       aTCRef);
    }
unknown's avatar
unknown committed
812
    return;
813 814
  }
  case GSN_TCINDXREF:{
unknown's avatar
unknown committed
815
    tFirstDataPtr = int2void(tFirstData);
816 817 818 819 820 821
    if (tFirstDataPtr == 0) goto InvalidSignal;

    tIndexOp = void2rec_iop(tFirstDataPtr);
    if (tIndexOp->checkMagicNumber() == 0) {
      tCon = tIndexOp->theNdbCon;
      if (tCon != NULL) {
unknown's avatar
unknown committed
822
	if (tCon->theSendStatus == NdbConnection::sendTC_OP) {
823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875
	  tReturnCode = tIndexOp->receiveTCINDXREF(aSignal);
	  if (tReturnCode != -1) {
	    completedTransaction(tCon);
	  }//if
	  return;
	}//if
      }//if
    }//if
    goto InvalidSignal;
    return;
  } 
  default:
    goto InvalidSignal;
  }//switch
  
  if (theWaiter.m_state == NO_WAIT) {
    // Wake up the thread waiting for response
    NdbCondition_Signal(theWaiter.m_condition);
  }//if
  return;

 InvalidSignal:
#ifdef VM_TRACE
  ndbout_c("Ndbif: Error Ndb::handleReceivedSignal "
	   "(GSN=%d, theWaiter.m_state=%d)"
	   " sender = (Block: %d Node: %d)",
	   tSignalNumber,
	   tWaitState,
	   refToBlock(aSignal->theSendersBlockRef),
	   refToNode(aSignal->theSendersBlockRef));
#endif
#ifdef NDB_NO_DROPPED_SIGNAL
  abort();
#endif
  
  return;
}//Ndb::handleReceivedSignal()


/*****************************************************************************
void completedTransaction(NdbConnection* aCon);

Remark:   One transaction has been completed.
          Remove it from send array and put it into the completed
          transaction array. Finally check if it is time to wake
          up a poller.
******************************************************************************/
void	
Ndb::completedTransaction(NdbConnection* aCon)
{
  Uint32 tTransArrayIndex = aCon->theTransArrayIndex;
  Uint32 tNoSentTransactions = theNoOfSentTransactions;
  Uint32 tNoCompletedTransactions = theNoOfCompletedTransactions;
unknown's avatar
unknown committed
876
  if ((tNoSentTransactions > 0) && (aCon->theListState == NdbConnection::InSendList) &&
877 878 879 880 881 882 883 884 885 886 887 888 889
      (tTransArrayIndex < tNoSentTransactions)) {
    NdbConnection* tMoveCon = theSentTransactionsArray[tNoSentTransactions - 1];

    theCompletedTransactionsArray[tNoCompletedTransactions] = aCon;
    aCon->theTransArrayIndex = tNoCompletedTransactions;
    if (tMoveCon != aCon) {
      tMoveCon->theTransArrayIndex = tTransArrayIndex;
      theSentTransactionsArray[tTransArrayIndex] = tMoveCon;
    }//if
    theSentTransactionsArray[tNoSentTransactions - 1] = NULL;
    theNoOfCompletedTransactions = tNoCompletedTransactions + 1;

    theNoOfSentTransactions = tNoSentTransactions - 1;
unknown's avatar
unknown committed
890
    aCon->theListState = NdbConnection::InCompletedList;
891 892 893 894 895 896 897 898
    aCon->handleExecuteCompletion();
    if ((theMinNoOfEventsToWakeUp != 0) &&
        (theNoOfCompletedTransactions >= theMinNoOfEventsToWakeUp)) {
      theMinNoOfEventsToWakeUp = 0;
      NdbCondition_Signal(theWaiter.m_condition);
      return;
    }//if
  } else {
unknown's avatar
unknown committed
899 900
    ndbout << "theNoOfSentTransactions = " << (int) theNoOfSentTransactions;
    ndbout << " theListState = " << (int) aCon->theListState;
901 902 903 904 905
    ndbout << " theTransArrayIndex = " << aCon->theTransArrayIndex;
    ndbout << endl << flush;
#ifdef VM_TRACE
    printState("completedTransaction abort");
    abort();
unknown's avatar
unknown committed
906
#endif
907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924
  }//if
}//Ndb::completedTransaction()

/*****************************************************************************
void reportCallback(NdbConnection** aCopyArray, Uint32 aNoOfCompletedTrans);

Remark:   Call the callback methods of the completed transactions.
******************************************************************************/
void	
Ndb::reportCallback(NdbConnection** aCopyArray, Uint32 aNoOfCompletedTrans)
{
  Uint32         i;
  if (aNoOfCompletedTrans > 0) {
    for (i = 0; i < aNoOfCompletedTrans; i++) {
      void* anyObject = aCopyArray[i]->theCallbackObject;
      NdbAsynchCallback aCallback = aCopyArray[i]->theCallbackFunction;
      int tResult = 0;
      if (aCallback != NULL) {
unknown's avatar
unknown committed
925
        if (aCopyArray[i]->theReturnStatus == NdbConnection::ReturnFailure) {
926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948
          tResult = -1;
        }//if
        (*aCallback)(tResult, aCopyArray[i], anyObject);
      }//if
    }//for
  }//if
}//Ndb::reportCallback()

/*****************************************************************************
Uint32 pollCompleted(NdbConnection** aCopyArray);

Remark:   Transfer the data from the completed transaction to a local array.
          This support is used by a number of the poll-methods.
******************************************************************************/
Uint32	
Ndb::pollCompleted(NdbConnection** aCopyArray)
{
  check_send_timeout();
  Uint32         i;
  Uint32 tNoCompletedTransactions = theNoOfCompletedTransactions;
  if (tNoCompletedTransactions > 0) {
    for (i = 0; i < tNoCompletedTransactions; i++) {
      aCopyArray[i] = theCompletedTransactionsArray[i];
unknown's avatar
unknown committed
949
      if (aCopyArray[i]->theListState != NdbConnection::InCompletedList) {
950
        ndbout << "pollCompleted error ";
unknown's avatar
unknown committed
951
        ndbout << (int) aCopyArray[i]->theListState << endl;
952 953 954
	abort();
      }//if
      theCompletedTransactionsArray[i] = NULL;
unknown's avatar
unknown committed
955
      aCopyArray[i]->theListState = NdbConnection::NotInList;
956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974
    }//for
  }//if
  theNoOfCompletedTransactions = 0;
  return tNoCompletedTransactions;
}//Ndb::pollCompleted()

void
Ndb::check_send_timeout()
{
  NDB_TICKS current_time = NdbTick_CurrentMillisecond();
  if (current_time - the_last_check_time > 1000) {
    the_last_check_time = current_time;
    Uint32 no_of_sent = theNoOfSentTransactions;
    for (Uint32 i = 0; i < no_of_sent; i++) {
      NdbConnection* a_con = theSentTransactionsArray[i];
      if ((current_time - a_con->theStartTransTime) >
          WAITFOR_RESPONSE_TIMEOUT) {
#ifdef VM_TRACE
        a_con->printState();
unknown's avatar
unknown committed
975 976 977 978
	Uint32 t1 = a_con->theTransactionId;
	Uint32 t2 = a_con->theTransactionId >> 32;
	ndbout_c("[%.8x %.8x]", t1, t2);
	abort();
979 980
#endif
        a_con->setOperationErrorCodeAbort(4012);
unknown's avatar
unknown committed
981 982
        a_con->theCommitStatus = NdbConnection::Aborted;
        a_con->theCompletionStatus = NdbConnection::CompletedFailure;
983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010
        a_con->handleExecuteCompletion();
        remove_sent_list(i);
        insert_completed_list(a_con);
        no_of_sent--;
        i--;
      }//if
    }//for
  }//if
}

void
Ndb::remove_sent_list(Uint32 list_index)
{
  Uint32 last_index = theNoOfSentTransactions - 1;
  if (list_index < last_index) {
    NdbConnection* t_con = theSentTransactionsArray[last_index];
    theSentTransactionsArray[list_index] = t_con;
  }//if
  theNoOfSentTransactions = last_index;
  theSentTransactionsArray[last_index] = 0;
}

Uint32
Ndb::insert_completed_list(NdbConnection* a_con)
{
  Uint32 no_of_comp = theNoOfCompletedTransactions;
  theCompletedTransactionsArray[no_of_comp] = a_con;
  theNoOfCompletedTransactions = no_of_comp + 1;
unknown's avatar
unknown committed
1011
  a_con->theListState = NdbConnection::InCompletedList;
1012 1013 1014 1015 1016 1017 1018 1019 1020 1021
  a_con->theTransArrayIndex = no_of_comp;
  return no_of_comp;
}

Uint32
Ndb::insert_sent_list(NdbConnection* a_con)
{
  Uint32 no_of_sent = theNoOfSentTransactions;
  theSentTransactionsArray[no_of_sent] = a_con;
  theNoOfSentTransactions = no_of_sent + 1;
unknown's avatar
unknown committed
1022
  a_con->theListState = NdbConnection::InSendList;
1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059
  a_con->theTransArrayIndex = no_of_sent;
  return no_of_sent;
}

/*****************************************************************************
void sendPrepTrans(int forceSend);

Remark: Send a batch of transactions prepared for sending to the NDB kernel.  
******************************************************************************/
void
Ndb::sendPrepTrans(int forceSend)
{
  // Always called when holding mutex on TransporterFacade
  /*
     We will send a list of transactions to the NDB kernel. Before
     sending we check the following.
     1) Node connected to is still alive
        Checked by both checking node status and node sequence
     2) Send buffer can handle the size of messages we are planning to send
        So far this is just a fake check but will soon be a real check
     When the connected node has failed we abort the transaction without
     responding anymore to the node since the kernel will clean up
     automatically.
     When sendBuffer cannot handle anymore messages then we will also abort
     transaction but by communicating to the kernel since it is still alive
     and we keep a small space for messages like that.
  */
  Uint32 i;
  TransporterFacade* tp = TransporterFacade::instance();
  Uint32 no_of_prep_trans = theNoOfPreparedTransactions;
  for (i = 0; i < no_of_prep_trans; i++) {
    NdbConnection * a_con = thePreparedTransactionsArray[i];
    thePreparedTransactionsArray[i] = NULL;
    Uint32 node_id = a_con->getConnectedNodeId();
    if ((tp->getNodeSequence(node_id) == a_con->theNodeSequence) &&
         tp->get_node_alive(node_id) ||
         (tp->get_node_stopping(node_id) && 
unknown's avatar
unknown committed
1060 1061 1062 1063
         ((a_con->theSendStatus == NdbConnection::sendABORT) ||
          (a_con->theSendStatus == NdbConnection::sendABORTfail) ||
          (a_con->theSendStatus == NdbConnection::sendCOMMITstate) ||
          (a_con->theSendStatus == NdbConnection::sendCompleted)))) {
1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094
      /*
      We will send if
      1) Node is alive and sequences are correct OR
      2) Node is stopping and we only want to commit or abort
      In a graceful stop situation we want to ensure quick aborts
      of all transactions and commits and thus we allow aborts and
      commits to continue but not normal operations.
      */
      if (tp->check_send_size(node_id, a_con->get_send_size())) {
        if (a_con->doSend() == 0) {
          NDB_TICKS current_time = NdbTick_CurrentMillisecond();
          a_con->theStartTransTime = current_time;
          continue;
        } else {
          /*
          Although all precautions we did not manage to send the operations
          Must have been a dropped connection on the transporter side.
          We don't expect to be able to continue using this connection so
          we will treat it as a node failure.
          */
          TRACE_DEBUG("Send problem even after checking node status");
        }//if
      } else {
        /*
        The send buffer is currently full or at least close to. We will
        not allow a send to continue. We will set the connection so that
        it is indicated that we need to abort the transaction. If we were
        trying to commit or abort and got a send buffer we will not try
        again and will thus set the state to Aborted to avoid a more or
        less eternal loop of tries.
        */
unknown's avatar
unknown committed
1095
        if (a_con->theSendStatus == NdbConnection::sendOperations) {
1096
          a_con->setOperationErrorCodeAbort(4021);
unknown's avatar
unknown committed
1097
          a_con->theCommitStatus = NdbConnection::NeedAbort;
1098 1099 1100
          TRACE_DEBUG("Send buffer full and sendOperations");
        } else {
          a_con->setOperationErrorCodeAbort(4026);
unknown's avatar
unknown committed
1101
          a_con->theCommitStatus = NdbConnection::Aborted;
1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117
          TRACE_DEBUG("Send buffer full, set state to Aborted");
        }//if
      }//if
    } else {
#ifdef VM_TRACE
      a_con->printState();
#endif
      if ((tp->getNodeSequence(node_id) == a_con->theNodeSequence) &&
          tp->get_node_stopping(node_id)) {
        /*
        The node we are connected to is currently in an early stopping phase
        of a graceful stop. We will not send the prepared transactions. We
        will simply refuse and let the application code handle the abort.
        */
        TRACE_DEBUG("Abort a transaction when stopping a node");
        a_con->setOperationErrorCodeAbort(4023);
unknown's avatar
unknown committed
1118
        a_con->theCommitStatus = NdbConnection::NeedAbort;
1119 1120 1121 1122 1123 1124 1125 1126 1127
      } else {
        /*
        The node is hard dead and we cannot continue. We will also release
        the connection to the free pool.
        */
        TRACE_DEBUG("The node was stone dead, inform about abort");
        a_con->setOperationErrorCodeAbort(4025);
        a_con->theReleaseOnClose = true;
        a_con->theTransactionIsStarted = false;
unknown's avatar
unknown committed
1128
        a_con->theCommitStatus = NdbConnection::Aborted;
1129 1130
      }//if
    }//if
unknown's avatar
unknown committed
1131
    a_con->theCompletionStatus = NdbConnection::CompletedFailure;
1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268
    a_con->handleExecuteCompletion();
    insert_completed_list(a_con);
  }//for
  theNoOfPreparedTransactions = 0;
  if (forceSend == 0) {
     tp->checkForceSend(theNdbBlockNumber);
  } else if (forceSend == 1) {
     tp->forceSend(theNdbBlockNumber);
  }//if
  return;
}//Ndb::sendPrepTrans()

/*****************************************************************************
void waitCompletedTransactions(int aMilliSecondsToWait, int noOfEventsToWaitFor);

Remark:   First send all prepared operations and then check if there are any
          transactions already completed. Do not wait for not completed
          transactions.
******************************************************************************/
void	
Ndb::waitCompletedTransactions(int aMilliSecondsToWait, 
			       int noOfEventsToWaitFor)
{
  theWaiter.m_state = NO_WAIT; 
  /**
   * theWaiter.m_state = NO_WAIT; 
   * To ensure no messup with synchronous node fail handling
   * (see ReportFailure)
   */
  int waitTime = aMilliSecondsToWait;
  NDB_TICKS maxTime = NdbTick_CurrentMillisecond() + (NDB_TICKS)waitTime;
  theMinNoOfEventsToWakeUp = noOfEventsToWaitFor;
  do {
    if (waitTime < 1000) waitTime = 1000;
    NdbCondition_WaitTimeout(theWaiter.m_condition,
			     (NdbMutex*)theWaiter.m_mutex,
			     waitTime);
    if (theNoOfCompletedTransactions >= (Uint32)noOfEventsToWaitFor) {
      break;
    }//if
    theMinNoOfEventsToWakeUp = noOfEventsToWaitFor;
    waitTime = (int)(maxTime - NdbTick_CurrentMillisecond());
  } while (waitTime > 0);
  return;
}//Ndb::waitCompletedTransactions()

/*****************************************************************************
void sendPreparedTransactions(int forceSend = 0);

Remark:   First send all prepared operations and then check if there are any
          transactions already completed. Do not wait for not completed
          transactions.
******************************************************************************/
void	
Ndb::sendPreparedTransactions(int forceSend)
{
  TransporterFacade::instance()->lock_mutex();
  sendPrepTrans(forceSend);
  TransporterFacade::instance()->unlock_mutex();
  return;
}//Ndb::sendPreparedTransactions()

/*****************************************************************************
int sendPollNdb(int aMillisecondNumber, int minNoOfEventsToWakeup = 1, int forceSend = 0);

Remark:   First send all prepared operations and then check if there are any
          transactions already completed. Wait for not completed
          transactions until the specified number have completed or until the
          timeout has occured. Timeout zero means no waiting time.
******************************************************************************/
int	
Ndb::sendPollNdb(int aMillisecondNumber, int minNoOfEventsToWakeup, int forceSend)
{
  NdbConnection* tConArray[1024];
  Uint32         tNoCompletedTransactions;

  //theCurrentConnectCounter = 0;
  //theCurrentConnectIndex++;
  TransporterFacade::instance()->lock_mutex();
  sendPrepTrans(forceSend);
  if ((minNoOfEventsToWakeup <= 0) ||
      ((Uint32)minNoOfEventsToWakeup > theNoOfSentTransactions)) {
    minNoOfEventsToWakeup = theNoOfSentTransactions;
  }//if
  if ((theNoOfCompletedTransactions < (Uint32)minNoOfEventsToWakeup) &&
      (aMillisecondNumber > 0)) {
    waitCompletedTransactions(aMillisecondNumber, minNoOfEventsToWakeup);
    tNoCompletedTransactions = pollCompleted(tConArray);
  } else {
    tNoCompletedTransactions = pollCompleted(tConArray);
  }//if
  TransporterFacade::instance()->unlock_mutex();
  reportCallback(tConArray, tNoCompletedTransactions);
  return tNoCompletedTransactions;
}//Ndb::sendPollNdb()

/*****************************************************************************
int pollNdb(int aMillisecondNumber, int minNoOfEventsToWakeup);

Remark:   Check if there are any transactions already completed. Wait for not
          completed transactions until the specified number have completed or
          until the timeout has occured. Timeout zero means no waiting time.
******************************************************************************/
int	
Ndb::pollNdb(int aMillisecondNumber, int minNoOfEventsToWakeup)
{
  NdbConnection* tConArray[1024];
  Uint32         tNoCompletedTransactions;

  //theCurrentConnectCounter = 0;
  //theCurrentConnectIndex++;
  TransporterFacade::instance()->lock_mutex();
  if ((minNoOfEventsToWakeup == 0) ||
      ((Uint32)minNoOfEventsToWakeup > theNoOfSentTransactions)) {
    minNoOfEventsToWakeup = theNoOfSentTransactions;
  }//if
  if ((theNoOfCompletedTransactions < (Uint32)minNoOfEventsToWakeup) &&
      (aMillisecondNumber > 0)) {
    waitCompletedTransactions(aMillisecondNumber, minNoOfEventsToWakeup);
    tNoCompletedTransactions = pollCompleted(tConArray);
  } else {
    tNoCompletedTransactions = pollCompleted(tConArray);
  }//if
  TransporterFacade::instance()->unlock_mutex();
  reportCallback(tConArray, tNoCompletedTransactions);
  return tNoCompletedTransactions;
}//Ndb::sendPollNdbWithoutWait()

/*****************************************************************************
int receiveOptimisedResponse();

Return:  0 - Response received
        -1 - Timeout occured waiting for response
        -2 - Node failure interupted wait for response

******************************************************************************/
int	
unknown's avatar
unknown committed
1269
Ndb::receiveResponse(int waitTime){
1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322
  int tResultCode;
  TransporterFacade::instance()->checkForceSend(theNdbBlockNumber);
  
  theWaiter.wait(waitTime);
  
  if(theWaiter.m_state == NO_WAIT) {
    tResultCode = 0;
  } else {

#ifdef VM_TRACE
    ndbout << "ERR: receiveResponse - theWaiter.m_state = ";
    ndbout << theWaiter.m_state << endl;
#endif

    if (theWaiter.m_state == WAIT_NODE_FAILURE){
      tResultCode = -2;
    } else {
      tResultCode = -1;
    }
    theWaiter.m_state = NO_WAIT;
  }
  return tResultCode;
}//Ndb::receiveResponse()

int
Ndb::sendRecSignal(Uint16 node_id,
		   Uint32 aWaitState,
		   NdbApiSignal* aSignal,
                   Uint32 conn_seq)
{
  /*
  In most situations 0 is returned.
  In error cases we have 5 different cases
  -1: Send ok, time out in waiting for reply
  -2: Node has failed
  -3: Send buffer not full, send failed yet
  -4: Send buffer full
  -5: Node is currently stopping
  */

  int return_code;
  TransporterFacade* tp = TransporterFacade::instance();
  Uint32 send_size = 1; // Always sends one signal only 
  tp->lock_mutex();
  // Protected area
  if ((tp->get_node_alive(node_id)) &&
      ((tp->getNodeSequence(node_id) == conn_seq) ||
       (conn_seq == 0))) {
    if (tp->check_send_size(node_id, send_size)) {
      return_code = tp->sendSignal(aSignal, node_id);
      if (return_code != -1) {
        theWaiter.m_node = node_id;
        theWaiter.m_state = aWaitState;
unknown's avatar
unknown committed
1323 1324 1325 1326
        return_code = receiveResponse();
      } else {
	return_code = -3;
      }
1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347
    } else {
      return_code = -4;
    }//if
  } else {
    if ((tp->get_node_stopping(node_id)) &&
        ((tp->getNodeSequence(node_id) == conn_seq) ||
         (conn_seq == 0))) {
      return_code = -5;
    } else {
      return_code = -2;
    }//if
  }//if
  tp->unlock_mutex();
  // End of protected area
  return return_code;
}//Ndb::sendRecSignal()

void
NdbConnection::sendTC_COMMIT_ACK(NdbApiSignal * aSignal,
				 Uint32 transId1, Uint32 transId2, 
				 Uint32 aTCRef){
unknown's avatar
unknown committed
1348 1349
#ifdef MARKER_TRACE
  ndbout_c("Sending TC_COMMIT_ACK(0x%.8x, 0x%.8x) to -> %d",
1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365
	   transId1,
	   transId2,
	   refToNode(aTCRef));
#endif  
  TransporterFacade *tp = TransporterFacade::instance();
  aSignal->theTrace                = TestOrd::TraceAPI;
  aSignal->theReceiversBlockNumber = DBTC;
  aSignal->theVerId_signalNumber   = GSN_TC_COMMIT_ACK;
  aSignal->theLength               = 2;

  Uint32 * dataPtr = aSignal->getDataPtrSend();
  dataPtr[0] = transId1;
  dataPtr[1] = transId2;

  tp->sendSignal(aSignal, refToNode(aTCRef));
}