Below is the list of changes that have just been committed into a local
5.1 repository of jonas. When jonas does a push these changes will
be propagated to the main repository and, within 24 hours after the
push, to the public repository.
For information on how to access the public repository
see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html
ChangeSet@stripped, 2006-09-26 13:58:34+02:00, jonas@stripped +4 -0
Merge perch.ndb.mysql.com:/home/jonas/src/50-work
into perch.ndb.mysql.com:/home/jonas/src/51-work
MERGE: 1.1810.1696.129
storage/ndb/include/ndbapi/NdbTransaction.hpp@stripped, 2006-09-26 13:58:32+02:00, jonas@stripped +1 -1
merge
MERGE: 1.41.3.2
storage/ndb/include/ndbapi/NdbTransaction.hpp@stripped, 2006-09-26 13:56:32+02:00, jonas@stripped +0 -0
Merge rename: ndb/include/ndbapi/NdbTransaction.hpp -> storage/ndb/include/ndbapi/NdbTransaction.hpp
storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp@stripped, 2006-09-26 13:56:32+02:00, jonas@stripped +0 -0
Auto merged
MERGE: 1.24.45.2
storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp@stripped, 2006-09-26 13:56:32+02:00, jonas@stripped +0 -0
Merge rename: ndb/src/kernel/blocks/dbdih/DbdihMain.cpp -> storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp
storage/ndb/src/ndbapi/NdbScanOperation.cpp@stripped, 2006-09-26 13:58:32+02:00, jonas@stripped +0 -1
merge
MERGE: 1.66.13.2
storage/ndb/src/ndbapi/NdbScanOperation.cpp@stripped, 2006-09-26 13:56:32+02:00, jonas@stripped +0 -0
Merge rename: ndb/src/ndbapi/NdbScanOperation.cpp -> storage/ndb/src/ndbapi/NdbScanOperation.cpp
storage/ndb/src/ndbapi/NdbTransaction.cpp@stripped, 2006-09-26 13:58:32+02:00, jonas@stripped +1 -38
merge
MERGE: 1.43.9.2
storage/ndb/src/ndbapi/NdbTransaction.cpp@stripped, 2006-09-26 13:56:32+02:00, jonas@stripped +0 -0
Merge rename: ndb/src/ndbapi/NdbTransaction.cpp -> storage/ndb/src/ndbapi/NdbTransaction.cpp
# This is a BitKeeper patch. What follows are the unified diffs for the
# set of deltas contained in the patch. The rest of the patch, the part
# that BitKeeper cares about, is below these diffs.
# User: jonas
# Host: perch.ndb.mysql.com
# Root: /home/jonas/src/51-work/RESYNC
--- 1.41.3.1/ndb/include/ndbapi/NdbTransaction.hpp 2006-09-26 13:58:38 +02:00
+++ 1.48/storage/ndb/include/ndbapi/NdbTransaction.hpp 2006-09-26 13:58:38 +02:00
@@ -140,6 +140,7 @@
friend class NdbIndexOperation;
friend class NdbIndexScanOperation;
friend class NdbBlob;
+ friend class ha_ndbcluster;
#endif
public:
@@ -795,7 +796,7 @@
bool theBlobFlag;
Uint8 thePendingBlobOps;
- static void sendTC_COMMIT_ACK(NdbApiSignal *,
+ static void sendTC_COMMIT_ACK(class TransporterFacade *, NdbApiSignal *,
Uint32 transId1, Uint32 transId2,
Uint32 aBlockRef);
--- 1.24.45.1/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp 2006-09-26 13:58:38 +02:00
+++ 1.80/storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp 2006-09-26 13:58:38 +02:00
@@ -67,6 +67,7 @@
#include <signaldata/CreateFragmentation.hpp>
#include <signaldata/LqhFrag.hpp>
#include <signaldata/FsOpenReq.hpp>
+#include <signaldata/DihFragCount.hpp>
#include <signaldata/DictLock.hpp>
#include <DebuggerNames.hpp>
@@ -610,6 +611,14 @@
checkWaitDropTabFailedLqh(signal, nodeId, tableId);
return;
}
+ case DihContinueB::ZTO_START_FRAGMENTS:
+ {
+ TakeOverRecordPtr takeOverPtr;
+ takeOverPtr.i = signal->theData[1];
+ ptrCheckGuard(takeOverPtr, MAX_NDB_NODES, takeOverRecord);
+ nr_start_fragments(signal, takeOverPtr);
+ return;
+ }
}//switch
ndbrequire(false);
@@ -640,9 +649,11 @@
c_copyGCISlave.m_expectedNextWord += CopyGCIReq::DATA_SIZE;
return;
}//if
-
+
+ Uint32 tmp= SYSFILE->m_restart_seq;
memcpy(sysfileData, cdata, sizeof(sysfileData));
-
+ SYSFILE->m_restart_seq = tmp;
+
c_copyGCISlave.m_copyReason = reason;
c_copyGCISlave.m_senderRef = signal->senderBlockRef();
c_copyGCISlave.m_senderData = copyGCI->anyData;
@@ -1053,7 +1064,7 @@
jamEntry();
const ndb_mgm_configuration_iterator * p =
- theConfiguration.getOwnConfigIterator();
+ m_ctx.m_config.getOwnConfigIterator();
ndbrequireErr(p != 0, NDBD_EXIT_INVALID_CONFIG);
initData();
@@ -1096,6 +1107,26 @@
return;
}//Dbdih::execSTART_FRAGCONF()
+void Dbdih::execSTART_FRAGREF(Signal* signal)
+{
+ jamEntry();
+
+ /**
+ * Kill starting node
+ */
+ Uint32 errCode = signal->theData[1];
+ Uint32 nodeId = signal->theData[2];
+
+ SystemError * const sysErr = (SystemError*)&signal->theData[0];
+ sysErr->errorCode = SystemError::StartFragRefError;
+ sysErr->errorRef = reference();
+ sysErr->data1 = errCode;
+ sysErr->data2 = 0;
+ sendSignal(calcNdbCntrBlockRef(nodeId), GSN_SYSTEM_ERROR, signal,
+ SystemError::SignalLength, JBB);
+ return;
+}//Dbdih::execSTART_FRAGCONF()
+
void Dbdih::execSTART_MEREF(Signal* signal)
{
jamEntry();
@@ -1139,7 +1170,7 @@
{
jamEntry();
cntrlblockref = signal->theData[0];
- if(theConfiguration.getInitialStart()){
+ if(m_ctx.m_config.getInitialStart()){
sendSignal(cntrlblockref, GSN_DIH_RESTARTREF, signal, 1, JBB);
} else {
readGciFileLab(signal);
@@ -1708,12 +1739,15 @@
*
* But dont copy lastCompletedGCI:s
*/
+ Uint32 key = SYSFILE->m_restart_seq;
Uint32 tempGCP[MAX_NDB_NODES];
for(i = 0; i < MAX_NDB_NODES; i++)
tempGCP[i] = SYSFILE->lastCompletedGCI[i];
for(i = 0; i < Sysfile::SYSFILE_SIZE32; i++)
sysfileData[i] = cdata[i];
+
+ SYSFILE->m_restart_seq = key;
for(i = 0; i < MAX_NDB_NODES; i++)
SYSFILE->lastCompletedGCI[i] = tempGCP[i];
@@ -1871,11 +1905,6 @@
ndbrequire(c_nodeStartMaster.startNode == Tnodeid);
ndbrequire(getNodeStatus(Tnodeid) == NodeRecord::STARTING);
- sendSTART_RECREQ(signal, Tnodeid);
-}//Dbdih::execSTART_MEREQ()
-
-void Dbdih::nodeRestartStartRecConfLab(Signal* signal)
-{
c_nodeStartMaster.blockLcp = true;
if ((c_lcpState.lcpStatus != LCP_STATUS_IDLE) &&
(c_lcpState.lcpStatus != LCP_TCGET)) {
@@ -2022,6 +2051,9 @@
signal->theData[0] = reference();
signal->theData[1] = c_nodeStartSlave.nodeId;
sendSignal(BACKUP_REF, GSN_INCL_NODEREQ, signal, 2, JBB);
+
+ // Suma will not send response to this for now, later...
+ sendSignal(SUMA_REF, GSN_INCL_NODEREQ, signal, 2, JBB);
return;
}//if
if (TstartNode_or_blockref == numberToRef(BACKUP, getOwnNodeId())){
@@ -2683,13 +2715,14 @@
return;
}//if
c_startToLock = takeOverPtrI;
+
+ takeOverPtr.p->toMasterStatus = TakeOverRecord::STARTING;
StartToReq * const req = (StartToReq *)&signal->theData[0];
req->userPtr = takeOverPtr.i;
req->userRef = reference();
req->startingNodeId = takeOverPtr.p->toStartingNode;
req->nodeTakenOver = takeOverPtr.p->toFailedNode;
req->nodeRestart = takeOverPtr.p->toNodeRestart;
- takeOverPtr.p->toMasterStatus = TakeOverRecord::STARTING;
sendLoopMacro(START_TOREQ, sendSTART_TOREQ);
}//Dbdih::sendStartTo()
@@ -2733,9 +2766,153 @@
CRASH_INSERTION(7134);
c_startToLock = RNIL;
+ if (takeOverPtr.p->toNodeRestart)
+ {
+ jam();
+ takeOverPtr.p->toMasterStatus = TakeOverRecord::STARTING_LOCAL_FRAGMENTS;
+ nr_start_fragments(signal, takeOverPtr);
+ return;
+ }
+
startNextCopyFragment(signal, takeOverPtr.i);
}//Dbdih::execSTART_TOCONF()
+void
+Dbdih::nr_start_fragments(Signal* signal,
+ TakeOverRecordPtr takeOverPtr)
+{
+ Uint32 loopCount = 0 ;
+ TabRecordPtr tabPtr;
+ while (loopCount++ < 100) {
+ tabPtr.i = takeOverPtr.p->toCurrentTabref;
+ if (tabPtr.i >= ctabFileSize) {
+ jam();
+ nr_run_redo(signal, takeOverPtr);
+ return;
+ }//if
+ ptrAss(tabPtr, tabRecord);
+ if (tabPtr.p->tabStatus != TabRecord::TS_ACTIVE){
+ jam();
+ takeOverPtr.p->toCurrentFragid = 0;
+ takeOverPtr.p->toCurrentTabref++;
+ continue;
+ }//if
+ Uint32 fragId = takeOverPtr.p->toCurrentFragid;
+ if (fragId >= tabPtr.p->totalfragments) {
+ jam();
+ takeOverPtr.p->toCurrentFragid = 0;
+ takeOverPtr.p->toCurrentTabref++;
+ continue;
+ }//if
+ FragmentstorePtr fragPtr;
+ getFragstore(tabPtr.p, fragId, fragPtr);
+ ReplicaRecordPtr loopReplicaPtr;
+ loopReplicaPtr.i = fragPtr.p->oldStoredReplicas;
+ while (loopReplicaPtr.i != RNIL) {
+ ptrCheckGuard(loopReplicaPtr, creplicaFileSize, replicaRecord);
+ if (loopReplicaPtr.p->procNode == takeOverPtr.p->toStartingNode) {
+ jam();
+ nr_start_fragment(signal, takeOverPtr, loopReplicaPtr);
+ break;
+ } else {
+ jam();
+ loopReplicaPtr.i = loopReplicaPtr.p->nextReplica;
+ }//if
+ }//while
+ takeOverPtr.p->toCurrentFragid++;
+ }//while
+ signal->theData[0] = DihContinueB::ZTO_START_FRAGMENTS;
+ signal->theData[1] = takeOverPtr.i;
+ sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
+}
+
+void
+Dbdih::nr_start_fragment(Signal* signal,
+ TakeOverRecordPtr takeOverPtr,
+ ReplicaRecordPtr replicaPtr)
+{
+ Uint32 i, j = 0;
+ Uint32 maxLcpId = 0;
+ Uint32 maxLcpIndex = ~0;
+
+ Uint32 restorableGCI = 0;
+
+ ndbout_c("tab: %d frag: %d replicaP->nextLcp: %d",
+ takeOverPtr.p->toCurrentTabref,
+ takeOverPtr.p->toCurrentFragid,
+ replicaPtr.p->nextLcp);
+
+ Uint32 idx = replicaPtr.p->nextLcp;
+ for(i = 0; i<MAX_LCP_STORED; i++, idx = nextLcpNo(idx))
+ {
+ ndbout_c("scanning idx: %d lcpId: %d", idx, replicaPtr.p->lcpId[idx]);
+ if (replicaPtr.p->lcpStatus[idx] == ZVALID)
+ {
+ ndbrequire(replicaPtr.p->lcpId[idx] > maxLcpId);
+ Uint32 startGci = replicaPtr.p->maxGciCompleted[idx];
+ Uint32 stopGci = replicaPtr.p->maxGciStarted[idx];
+ for (;j < replicaPtr.p->noCrashedReplicas; j++)
+ {
+ ndbout_c("crashed replica: %d(%d) replicaLastGci: %d",
+ j,
+ replicaPtr.p->noCrashedReplicas,
+ replicaPtr.p->replicaLastGci[j]);
+ if (replicaPtr.p->replicaLastGci[j] > stopGci)
+ {
+ maxLcpId = replicaPtr.p->lcpId[idx];
+ maxLcpIndex = idx;
+ restorableGCI = replicaPtr.p->replicaLastGci[j];
+ break;
+ }
+ }
+ }
+ }
+
+ if (maxLcpIndex == ~0)
+ {
+ ndbout_c("Didnt find any LCP for node: %d tab: %d frag: %d",
+ takeOverPtr.p->toStartingNode,
+ takeOverPtr.p->toCurrentTabref,
+ takeOverPtr.p->toCurrentFragid);
+ replicaPtr.p->lcpIdStarted = 0;
+ }
+ else
+ {
+ ndbout_c("Found LCP: %d(%d) maxGciStarted: %d maxGciCompleted: %d restorable: %d(%d) newestRestorableGCI: %d",
+ maxLcpId,
+ maxLcpIndex,
+ replicaPtr.p->maxGciStarted[maxLcpIndex],
+ replicaPtr.p->maxGciCompleted[maxLcpIndex],
+ restorableGCI,
+ SYSFILE->lastCompletedGCI[takeOverPtr.p->toStartingNode],
+ SYSFILE->newestRestorableGCI);
+
+ replicaPtr.p->lcpIdStarted = restorableGCI;
+ BlockReference ref = calcLqhBlockRef(takeOverPtr.p->toStartingNode);
+ StartFragReq *req = (StartFragReq *)signal->getDataPtrSend();
+ req->userPtr = 0;
+ req->userRef = reference();
+ req->lcpNo = maxLcpIndex;
+ req->lcpId = maxLcpId;
+ req->tableId = takeOverPtr.p->toCurrentTabref;
+ req->fragId = takeOverPtr.p->toCurrentFragid;
+ req->noOfLogNodes = 1;
+ req->lqhLogNode[0] = takeOverPtr.p->toStartingNode;
+ req->startGci[0] = replicaPtr.p->maxGciCompleted[maxLcpIndex];
+ req->lastGci[0] = restorableGCI;
+ sendSignal(ref, GSN_START_FRAGREQ, signal,
+ StartFragReq::SignalLength, JBB);
+ }
+}
+
+void
+Dbdih::nr_run_redo(Signal* signal, TakeOverRecordPtr takeOverPtr)
+{
+ takeOverPtr.p->toCurrentTabref = 0;
+ takeOverPtr.p->toCurrentFragid = 0;
+ sendSTART_RECREQ(signal, takeOverPtr.p->toStartingNode);
+}
+
void Dbdih::initStartTakeOver(const StartToReq * req,
TakeOverRecordPtr takeOverPtr)
{
@@ -3068,6 +3245,14 @@
/*---------------------------------------------------------------------- */
FragmentstorePtr fragPtr;
getFragstore(tabPtr.p, fragId, fragPtr);
+ Uint32 gci = 0;
+ if (takeOverPtr.p->toNodeRestart)
+ {
+ ReplicaRecordPtr replicaPtr;
+ findReplica(replicaPtr, fragPtr.p, takeOverPtr.p->toStartingNode, true);
+ gci = replicaPtr.p->lcpIdStarted;
+ replicaPtr.p->lcpIdStarted = 0;
+ }
takeOverPtr.p->toMasterStatus = TakeOverRecord::COPY_FRAG;
BlockReference ref = calcLqhBlockRef(takeOverPtr.p->toCopyNode);
CopyFragReq * const copyFragReq = (CopyFragReq *)&signal->theData[0];
@@ -3078,6 +3263,7 @@
copyFragReq->nodeId = takeOverPtr.p->toStartingNode;
copyFragReq->schemaVersion = tabPtr.p->schemaVersion;
copyFragReq->distributionKey = fragPtr.p->distributionKey;
+ copyFragReq->gci = gci;
copyFragReq->nodeCount = extractNodeInfo(fragPtr.p,
copyFragReq->nodeList);
sendSignal(ref, GSN_COPY_FRAGREQ, signal,
@@ -3602,6 +3788,7 @@
/* WE ALSO COPY TO OUR OWN NODE. TO ENABLE US TO DO THIS PROPERLY WE */
/* START BY CLOSING THIS FILE. */
/* ----------------------------------------------------------------------- */
+ globalData.m_restart_seq = ++SYSFILE->m_restart_seq;
closeFile(signal, filePtr);
filePtr.p->reqStatus = FileRecord::CLOSING_GCP;
}//Dbdih::readingGcpLab()
@@ -4120,6 +4307,8 @@
Uint32 takeOverPtrI)
{
jam();
+ ndbout_c("checkTakeOverInMasterStartNodeFailure %x",
+ takeOverPtrI);
if (takeOverPtrI == RNIL) {
jam();
return;
@@ -4133,6 +4322,9 @@
takeOverPtr.i = takeOverPtrI;
ptrCheckGuard(takeOverPtr, MAX_NDB_NODES, takeOverRecord);
+ ndbout_c("takeOverPtr.p->toMasterStatus: %x",
+ takeOverPtr.p->toMasterStatus);
+
bool ok = false;
switch (takeOverPtr.p->toMasterStatus) {
case TakeOverRecord::IDLE:
@@ -4241,6 +4433,13 @@
//-----------------------------------------------------------------------
endTakeOver(takeOverPtr.i);
break;
+
+ case TakeOverRecord::STARTING_LOCAL_FRAGMENTS:
+ ok = true;
+ jam();
+ endTakeOver(takeOverPtr.i);
+ break;
+
/**
* The following are states that it should not be possible to "be" in
*/
@@ -5659,11 +5858,9 @@
#endif
}
- bool ok = false;
MasterLCPConf::State lcpState;
switch (c_lcpState.lcpStatus) {
case LCP_STATUS_IDLE:
- ok = true;
jam();
/*------------------------------------------------*/
/* LOCAL CHECKPOINT IS CURRENTLY NOT ACTIVE */
@@ -5674,7 +5871,6 @@
lcpState = MasterLCPConf::LCP_STATUS_IDLE;
break;
case LCP_STATUS_ACTIVE:
- ok = true;
jam();
/*--------------------------------------------------*/
/* COPY OF RESTART INFORMATION HAS BEEN */
@@ -5683,7 +5879,6 @@
lcpState = MasterLCPConf::LCP_STATUS_ACTIVE;
break;
case LCP_TAB_COMPLETED:
- ok = true;
jam();
/*--------------------------------------------------------*/
/* ALL LCP_REPORT'S HAVE BEEN COMPLETED FOR */
@@ -5693,7 +5888,6 @@
lcpState = MasterLCPConf::LCP_TAB_COMPLETED;
break;
case LCP_TAB_SAVED:
- ok = true;
jam();
/*--------------------------------------------------------*/
/* ALL LCP_REPORT'S HAVE BEEN COMPLETED FOR */
@@ -5717,15 +5911,15 @@
break;
case LCP_COPY_GCI:
case LCP_INIT_TABLES:
- ok = true;
/**
* These two states are handled by if statements above
*/
ndbrequire(false);
lcpState= MasterLCPConf::LCP_STATUS_IDLE; // remove warning
break;
+ default:
+ ndbrequire(false);
}//switch
- ndbrequire(ok);
Uint32 failedNodeId = c_lcpState.m_MASTER_LCPREQ_FailedNodeId;
MasterLCPConf * const conf = (MasterLCPConf *)&signal->theData[0];
@@ -6323,96 +6517,147 @@
3.7.1 A D D T A B L E M A I N L Y
***************************************
*/
-void Dbdih::execCREATE_FRAGMENTATION_REQ(Signal * signal){
+
+static inline void inc_node_or_group(Uint32 &node, Uint32 max_node)
+{
+ Uint32 next = node + 1;
+ node = (next == max_node ? 0 : next);
+}
+
+/*
+ Spread fragments in backwards compatible mode
+*/
+static void set_default_node_groups(Signal *signal, Uint32 noFrags)
+{
+ Uint16 *node_group_array = (Uint16*)&signal->theData[25];
+ Uint32 i;
+ node_group_array[0] = 0;
+ for (i = 1; i < noFrags; i++)
+ node_group_array[i] = UNDEF_NODEGROUP;
+}
+void Dbdih::execCREATE_FRAGMENTATION_REQ(Signal * signal)
+{
+ Uint16 node_group_id[MAX_NDB_PARTITIONS];
jamEntry();
CreateFragmentationReq * const req =
(CreateFragmentationReq*)signal->getDataPtr();
const Uint32 senderRef = req->senderRef;
const Uint32 senderData = req->senderData;
- const Uint32 fragmentNode = req->fragmentNode;
- const Uint32 fragmentType = req->fragmentationType;
- //const Uint32 fragmentCount = req->noOfFragments;
+ Uint32 noOfFragments = req->noOfFragments;
+ const Uint32 fragType = req->fragmentationType;
const Uint32 primaryTableId = req->primaryTableId;
Uint32 err = 0;
do {
- Uint32 noOfFragments = 0;
- Uint32 noOfReplicas = cnoReplicas;
- switch(fragmentType){
- case DictTabInfo::AllNodesSmallTable:
- jam();
- noOfFragments = csystemnodes;
- break;
- case DictTabInfo::AllNodesMediumTable:
- jam();
- noOfFragments = 2 * csystemnodes;
- break;
- case DictTabInfo::AllNodesLargeTable:
- jam();
- noOfFragments = 4 * csystemnodes;
- break;
- case DictTabInfo::SingleFragment:
- jam();
- noOfFragments = 1;
- break;
-#if 0
- case DictTabInfo::SpecifiedFragmentCount:
- noOfFragments = (fragmentCount == 0 ? 1 : (fragmentCount + 1)/ 2);
- break;
-#endif
- default:
- jam();
- err = CreateFragmentationRef::InvalidFragmentationType;
- break;
- }
- if(err)
- break;
-
NodeGroupRecordPtr NGPtr;
TabRecordPtr primTabPtr;
+ Uint32 count = 2;
+ Uint16 noOfReplicas = cnoReplicas;
+ Uint16 *fragments = (Uint16*)(signal->theData+25);
if (primaryTableId == RNIL) {
- if(fragmentNode == 0){
- jam();
- NGPtr.i = 0;
- if(noOfFragments < csystemnodes)
- {
- NGPtr.i = c_nextNodeGroup;
- c_nextNodeGroup = (NGPtr.i + 1 == cnoOfNodeGroups ? 0 : NGPtr.i + 1);
- }
- } else if(! (fragmentNode < MAX_NDB_NODES)) {
- jam();
- err = CreateFragmentationRef::InvalidNodeId;
- } else {
- jam();
- const Uint32 stat = Sysfile::getNodeStatus(fragmentNode,
- SYSFILE->nodeStatus);
- switch (stat) {
- case Sysfile::NS_Active:
- case Sysfile::NS_ActiveMissed_1:
- case Sysfile::NS_ActiveMissed_2:
- case Sysfile::NS_TakeOver:
+ jam();
+ switch ((DictTabInfo::FragmentType)fragType)
+ {
+ /*
+ Backward compatability and for all places in code not changed.
+ */
+ case DictTabInfo::AllNodesSmallTable:
jam();
+ noOfFragments = csystemnodes;
+ set_default_node_groups(signal, noOfFragments);
break;
- case Sysfile::NS_NotActive_NotTakenOver:
+ case DictTabInfo::AllNodesMediumTable:
jam();
+ noOfFragments = 2 * csystemnodes;
+ set_default_node_groups(signal, noOfFragments);
break;
- case Sysfile::NS_HotSpare:
+ case DictTabInfo::AllNodesLargeTable:
jam();
- case Sysfile::NS_NotDefined:
+ noOfFragments = 4 * csystemnodes;
+ set_default_node_groups(signal, noOfFragments);
+ break;
+ case DictTabInfo::SingleFragment:
+ jam();
+ noOfFragments = 1;
+ set_default_node_groups(signal, noOfFragments);
+ break;
+ case DictTabInfo::DistrKeyHash:
jam();
+ case DictTabInfo::DistrKeyLin:
+ jam();
+ if (noOfFragments == 0)
+ {
+ jam();
+ noOfFragments = csystemnodes;
+ set_default_node_groups(signal, noOfFragments);
+ }
+ break;
default:
jam();
- err = CreateFragmentationRef::InvalidNodeType;
+ if (noOfFragments == 0)
+ {
+ jam();
+ err = CreateFragmentationRef::InvalidFragmentationType;
+ }
break;
+ }
+ if (err)
+ break;
+ /*
+ When we come here the the exact partition is specified
+ and there is an array of node groups sent along as well.
+ */
+ memcpy(&node_group_id[0], &signal->theData[25], 2 * noOfFragments);
+ Uint16 next_replica_node[MAX_NDB_NODES];
+ memset(next_replica_node,0,sizeof(next_replica_node));
+ Uint32 default_node_group= c_nextNodeGroup;
+ for(Uint32 fragNo = 0; fragNo < noOfFragments; fragNo++)
+ {
+ jam();
+ NGPtr.i = node_group_id[fragNo];
+ if (NGPtr.i == UNDEF_NODEGROUP)
+ {
+ jam();
+ NGPtr.i = default_node_group;
}
- if(err)
+ if (NGPtr.i > cnoOfNodeGroups)
+ {
+ jam();
+ err = CreateFragmentationRef::InvalidNodeGroup;
break;
- NGPtr.i = Sysfile::getNodeGroup(fragmentNode,
- SYSFILE->nodeGroups);
+ }
+ ptrCheckGuard(NGPtr, MAX_NDB_NODES, nodeGroupRecord);
+ const Uint32 max = NGPtr.p->nodeCount;
+
+ fragments[count++] = c_nextLogPart++; // Store logpart first
+ Uint32 tmp= next_replica_node[NGPtr.i];
+ for(Uint32 replicaNo = 0; replicaNo < noOfReplicas; replicaNo++)
+ {
+ jam();
+ const Uint16 nodeId = NGPtr.p->nodesInGroup[tmp];
+ fragments[count++]= nodeId;
+ inc_node_or_group(tmp, max);
+ }
+ inc_node_or_group(tmp, max);
+ next_replica_node[NGPtr.i]= tmp;
+
+ /**
+ * Next node group for next fragment
+ */
+ inc_node_or_group(default_node_group, cnoOfNodeGroups);
+ }
+ if (err)
+ {
+ jam();
break;
}
+ else
+ {
+ jam();
+ c_nextNodeGroup = default_node_group;
+ }
} else {
if (primaryTableId >= ctabFileSize) {
jam();
@@ -6426,48 +6671,14 @@
err = CreateFragmentationRef::InvalidPrimaryTable;
break;
}
- if (noOfFragments != primTabPtr.p->totalfragments) {
- jam();
- err = CreateFragmentationRef::InvalidFragmentationType;
- break;
- }
- }
-
- Uint32 count = 2;
- Uint16 *fragments = (Uint16*)(signal->theData+25);
- if (primaryTableId == RNIL) {
- jam();
- Uint8 next_replica_node[MAX_NDB_NODES];
- memset(next_replica_node,0,sizeof(next_replica_node));
- for(Uint32 fragNo = 0; fragNo<noOfFragments; fragNo++){
- jam();
- ptrCheckGuard(NGPtr, MAX_NDB_NODES, nodeGroupRecord);
- const Uint32 max = NGPtr.p->nodeCount;
-
- Uint32 tmp= next_replica_node[NGPtr.i];
- for(Uint32 replicaNo = 0; replicaNo<noOfReplicas; replicaNo++)
- {
- jam();
- const Uint32 nodeId = NGPtr.p->nodesInGroup[tmp++];
- fragments[count++] = nodeId;
- tmp = (tmp >= max ? 0 : tmp);
- }
- tmp++;
- next_replica_node[NGPtr.i]= (tmp >= max ? 0 : tmp);
-
- /**
- * Next node group for next fragment
- */
- NGPtr.i++;
- NGPtr.i = (NGPtr.i == cnoOfNodeGroups ? 0 : NGPtr.i);
- }
- } else {
+ noOfFragments= primTabPtr.p->totalfragments;
for (Uint32 fragNo = 0;
- fragNo < primTabPtr.p->totalfragments; fragNo++) {
+ fragNo < noOfFragments; fragNo++) {
jam();
FragmentstorePtr fragPtr;
ReplicaRecordPtr replicaPtr;
getFragstore(primTabPtr.p, fragNo, fragPtr);
+ fragments[count++] = c_nextLogPart++;
fragments[count++] = fragPtr.p->preferredPrimary;
for (replicaPtr.i = fragPtr.p->storedReplicas;
replicaPtr.i != RNIL;
@@ -6476,9 +6687,9 @@
ptrCheckGuard(replicaPtr, creplicaFileSize, replicaRecord);
if (replicaPtr.p->procNode != fragPtr.p->preferredPrimary) {
jam();
- fragments[count++] = replicaPtr.p->procNode;
- }//if
- }//for
+ fragments[count++]= replicaPtr.p->procNode;
+ }
+ }
for (replicaPtr.i = fragPtr.p->oldStoredReplicas;
replicaPtr.i != RNIL;
replicaPtr.i = replicaPtr.p->nextReplica) {
@@ -6486,25 +6697,26 @@
ptrCheckGuard(replicaPtr, creplicaFileSize, replicaRecord);
if (replicaPtr.p->procNode != fragPtr.p->preferredPrimary) {
jam();
- fragments[count++] = replicaPtr.p->procNode;
- }//if
- }//for
+ fragments[count++]= replicaPtr.p->procNode;
+ }
+ }
}
}
- ndbrequire(count == (2 + noOfReplicas * noOfFragments));
+ ndbrequire(count == (2U + (1 + noOfReplicas) * noOfFragments));
CreateFragmentationConf * const conf =
(CreateFragmentationConf*)signal->getDataPtrSend();
conf->senderRef = reference();
conf->senderData = senderData;
- conf->noOfReplicas = noOfReplicas;
- conf->noOfFragments = noOfFragments;
+ conf->noOfReplicas = (Uint32)noOfReplicas;
+ conf->noOfFragments = (Uint32)noOfFragments;
- fragments[0] = noOfReplicas;
- fragments[1] = noOfFragments;
+ fragments[0]= noOfReplicas;
+ fragments[1]= noOfFragments;
if(senderRef != 0)
{
+ jam();
LinearSectionPtr ptr[3];
ptr[0].p = (Uint32*)&fragments[0];
ptr[0].sz = (count + 1) / 2;
@@ -6516,33 +6728,17 @@
ptr,
1);
}
- else
- {
- // Execute direct
- signal->theData[0] = 0;
- }
+ // Always ACK/NACK (here ACK)
+ signal->theData[0] = 0;
return;
} while(false);
-
- if(senderRef != 0)
- {
- CreateFragmentationRef * const ref =
- (CreateFragmentationRef*)signal->getDataPtrSend();
- ref->senderRef = reference();
- ref->senderData = senderData;
- ref->errorCode = err;
- sendSignal(senderRef, GSN_CREATE_FRAGMENTATION_REF, signal,
- CreateFragmentationRef::SignalLength, JBB);
- }
- else
- {
- // Execute direct
- signal->theData[0] = err;
- }
+ // Always ACK/NACK (here NACK)
+ signal->theData[0] = err;
}
void Dbdih::execDIADDTABREQ(Signal* signal)
{
+ Uint32 fragType;
jamEntry();
DiAddTabReq * const req = (DiAddTabReq*)signal->getDataPtr();
@@ -6567,6 +6763,7 @@
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
tabPtr.p->connectrec = connectPtr.i;
tabPtr.p->tableType = req->tableType;
+ fragType= req->fragType;
tabPtr.p->schemaVersion = req->schemaVersion;
tabPtr.p->primaryTableId = req->primaryTableId;
@@ -6603,9 +6800,33 @@
/*%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%*/
tabPtr.p->tabStatus = TabRecord::TS_CREATING;
tabPtr.p->storedTable = req->storedTable;
- tabPtr.p->method = TabRecord::HASH;
tabPtr.p->kvalue = req->kValue;
+ switch ((DictTabInfo::FragmentType)fragType)
+ {
+ case DictTabInfo::AllNodesSmallTable:
+ case DictTabInfo::AllNodesMediumTable:
+ case DictTabInfo::AllNodesLargeTable:
+ case DictTabInfo::SingleFragment:
+ jam();
+ case DictTabInfo::DistrKeyLin:
+ jam();
+ tabPtr.p->method= TabRecord::LINEAR_HASH;
+ break;
+ case DictTabInfo::DistrKeyHash:
+ case DictTabInfo::DistrKeyUniqueHashIndex:
+ case DictTabInfo::DistrKeyOrderedIndex:
+ jam();
+ tabPtr.p->method= TabRecord::NORMAL_HASH;
+ break;
+ case DictTabInfo::UserDefined:
+ jam();
+ tabPtr.p->method= TabRecord::USER_DEFINED;
+ break;
+ default:
+ ndbrequire(false);
+ }
+
union {
Uint16 fragments[2 + MAX_FRAG_PER_NODE*MAX_REPLICAS*MAX_NDB_NODES];
Uint32 align;
@@ -6654,7 +6875,9 @@
FragmentstorePtr fragPtr;
Uint32 activeIndex = 0;
getFragstore(tabPtr.p, fragId, fragPtr);
+ fragPtr.p->m_log_part_id = fragments[index++];
fragPtr.p->preferredPrimary = fragments[index];
+
for (Uint32 i = 0; i<noReplicas; i++) {
const Uint32 nodeId = fragments[index++];
ReplicaRecordPtr replicaPtr;
@@ -6699,9 +6922,9 @@
jam();
const Uint32 fragCount = tabPtr.p->totalfragments;
ReplicaRecordPtr replicaPtr; replicaPtr.i = RNIL;
+ FragmentstorePtr fragPtr;
for(; fragId<fragCount; fragId++){
jam();
- FragmentstorePtr fragPtr;
getFragstore(tabPtr.p, fragId, fragPtr);
replicaPtr.i = fragPtr.p->storedReplicas;
@@ -6759,6 +6982,7 @@
req->nodeId = getOwnNodeId();
req->totalFragments = fragCount;
req->startGci = SYSFILE->newestRestorableGCI;
+ req->logPartId = fragPtr.p->m_log_part_id;
sendSignal(DBDICT_REF, GSN_ADD_FRAGREQ, signal,
AddFragReq::SignalLength, JBB);
return;
@@ -7040,17 +7264,40 @@
tabPtr.i = req->tableId;
Uint32 hashValue = req->hashValue;
Uint32 ttabFileSize = ctabFileSize;
+ Uint32 fragId;
+ DiGetNodesConf * const conf = (DiGetNodesConf *)&signal->theData[0];
TabRecord* regTabDesc = tabRecord;
jamEntry();
ptrCheckGuard(tabPtr, ttabFileSize, regTabDesc);
- Uint32 fragId = hashValue & tabPtr.p->mask;
- ndbrequire(tabPtr.p->tabStatus == TabRecord::TS_ACTIVE);
- if (fragId < tabPtr.p->hashpointer) {
+ if (tabPtr.p->method == TabRecord::LINEAR_HASH)
+ {
jam();
- fragId = hashValue & ((tabPtr.p->mask << 1) + 1);
- }//if
+ fragId = hashValue & tabPtr.p->mask;
+ ndbrequire(tabPtr.p->tabStatus == TabRecord::TS_ACTIVE);
+ if (fragId < tabPtr.p->hashpointer) {
+ jam();
+ fragId = hashValue & ((tabPtr.p->mask << 1) + 1);
+ }//if
+ }
+ else if (tabPtr.p->method == TabRecord::NORMAL_HASH)
+ {
+ jam();
+ fragId= hashValue % tabPtr.p->totalfragments;
+ }
+ else
+ {
+ jam();
+ ndbassert(tabPtr.p->method == TabRecord::USER_DEFINED);
+ fragId= hashValue;
+ if (fragId >= tabPtr.p->totalfragments)
+ {
+ jam();
+ conf->zero= 1; //Indicate error;
+ signal->theData[1]= ZUNDEFINED_FRAGMENT_ERROR;
+ return;
+ }
+ }
getFragstore(tabPtr.p, fragId, fragPtr);
- DiGetNodesConf * const conf = (DiGetNodesConf *)&signal->theData[0];
Uint32 nodeCount = extractNodeInfo(fragPtr.p, conf->nodes);
Uint32 sig2 = (nodeCount - 1) +
(fragPtr.p->distributionKey << 16);
@@ -7217,39 +7464,66 @@
void Dbdih::execDI_FCOUNTREQ(Signal* signal)
{
+ DihFragCountReq * const req = (DihFragCountReq*)signal->getDataPtr();
ConnectRecordPtr connectPtr;
TabRecordPtr tabPtr;
+ const BlockReference senderRef = signal->senderBlockRef();
+ const Uint32 senderData = req->m_senderData;
jamEntry();
- connectPtr.i = signal->theData[0];
- tabPtr.i = signal->theData[1];
+ connectPtr.i = req->m_connectionData;
+ tabPtr.i = req->m_tableRef;
ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
- ndbrequire(tabPtr.p->tabStatus == TabRecord::TS_ACTIVE);
+ if (tabPtr.p->tabStatus != TabRecord::TS_ACTIVE)
+ {
+ DihFragCountRef* ref = (DihFragCountRef*)signal->getDataPtrSend();
+ //connectPtr.i == RNIL -> question without connect record
+ if(connectPtr.i == RNIL)
+ ref->m_connectionData = RNIL;
+ else
+ ref->m_connectionData = connectPtr.p->userpointer;
+ ref->m_tableRef = tabPtr.i;
+ ref->m_senderData = senderData;
+ ref->m_error = DihFragCountRef::ErroneousTableState;
+ ref->m_tableStatus = tabPtr.p->tabStatus;
+ sendSignal(senderRef, GSN_DI_FCOUNTREF, signal,
+ DihFragCountRef::SignalLength, JBB);
+ return;
+ }
if(connectPtr.i != RNIL){
ptrCheckGuard(connectPtr, cconnectFileSize, connectRecord);
if (connectPtr.p->connectState == ConnectRecord::INUSE) {
jam();
- signal->theData[0] = connectPtr.p->userpointer;
- signal->theData[1] = tabPtr.p->totalfragments;
- sendSignal(connectPtr.p->userblockref, GSN_DI_FCOUNTCONF, signal,2, JBB);
- return;
- }//if
- signal->theData[0] = connectPtr.p->userpointer;
- signal->theData[1] = ZERRONOUSSTATE;
- sendSignal(connectPtr.p->userblockref, GSN_DI_FCOUNTREF, signal, 2, JBB);
+ DihFragCountConf* conf = (DihFragCountConf*)signal->getDataPtrSend();
+ conf->m_connectionData = connectPtr.p->userpointer;
+ conf->m_tableRef = tabPtr.i;
+ conf->m_senderData = senderData;
+ conf->m_fragmentCount = tabPtr.p->totalfragments;
+ conf->m_noOfBackups = tabPtr.p->noOfBackups;
+ sendSignal(connectPtr.p->userblockref, GSN_DI_FCOUNTCONF, signal,
+ DihFragCountConf::SignalLength, JBB);
+ return;
+ }//if
+ DihFragCountRef* ref = (DihFragCountRef*)signal->getDataPtrSend();
+ ref->m_connectionData = connectPtr.p->userpointer;
+ ref->m_tableRef = tabPtr.i;
+ ref->m_senderData = senderData;
+ ref->m_error = DihFragCountRef::ErroneousTableState;
+ ref->m_tableStatus = tabPtr.p->tabStatus;
+ sendSignal(connectPtr.p->userblockref, GSN_DI_FCOUNTREF, signal,
+ DihFragCountRef::SignalLength, JBB);
return;
}//if
-
+ DihFragCountConf* conf = (DihFragCountConf*)signal->getDataPtrSend();
//connectPtr.i == RNIL -> question without connect record
- const Uint32 senderData = signal->theData[2];
- const BlockReference senderRef = signal->senderBlockRef();
- signal->theData[0] = RNIL;
- signal->theData[1] = tabPtr.p->totalfragments;
- signal->theData[2] = tabPtr.i;
- signal->theData[3] = senderData;
- signal->theData[4] = tabPtr.p->noOfBackups;
- sendSignal(senderRef, GSN_DI_FCOUNTCONF, signal, 5, JBB);
+ conf->m_connectionData = RNIL;
+ conf->m_tableRef = tabPtr.i;
+ conf->m_senderData = senderData;
+ conf->m_fragmentCount = tabPtr.p->totalfragments;
+ conf->m_noOfBackups = tabPtr.p->noOfBackups;
+ sendSignal(senderRef, GSN_DI_FCOUNTCONF, signal,
+ DihFragCountConf::SignalLength, JBB);
}//Dbdih::execDI_FCOUNTREQ()
void Dbdih::execDIGETPRIMREQ(Signal* signal)
@@ -8023,6 +8297,15 @@
if (reason == CopyGCIReq::GLOBAL_CHECKPOINT) {
jam();
cgcpParticipantState = GCP_PARTICIPANT_READY;
+
+ SubGcpCompleteRep * const rep = (SubGcpCompleteRep*)signal->getDataPtr();
+ rep->gci = coldgcp;
+ sendSignal(SUMA_REF, GSN_SUB_GCP_COMPLETE_REP, signal,
+ SubGcpCompleteRep::SignalLength, JBB);
+
+ EXECUTE_DIRECT(LGMAN, GSN_SUB_GCP_COMPLETE_REP, signal,
+ SubGcpCompleteRep::SignalLength);
+ jamEntry();
}
jam();
@@ -8600,8 +8883,7 @@
rf.rwfTabPtr.p->hashpointer = readPageWord(&rf);
rf.rwfTabPtr.p->kvalue = readPageWord(&rf);
rf.rwfTabPtr.p->mask = readPageWord(&rf);
- ndbrequire(readPageWord(&rf) == TabRecord::HASH);
- rf.rwfTabPtr.p->method = TabRecord::HASH;
+ rf.rwfTabPtr.p->method = (TabRecord::Method)readPageWord(&rf);
/* ---------------------------------- */
/* Type of table, 2 = temporary table */
/* ---------------------------------- */
@@ -8695,7 +8977,7 @@
writePageWord(&wf, tabPtr.p->hashpointer);
writePageWord(&wf, tabPtr.p->kvalue);
writePageWord(&wf, tabPtr.p->mask);
- writePageWord(&wf, TabRecord::HASH);
+ writePageWord(&wf, tabPtr.p->method);
writePageWord(&wf, tabPtr.p->storedTable);
signal->theData[0] = DihContinueB::ZPACK_FRAG_INTO_PAGES;
@@ -8796,6 +9078,80 @@
/*****************************************************************************/
/* ********** START FRAGMENT MODULE *************/
/*****************************************************************************/
+void
+Dbdih::dump_replica_info()
+{
+ TabRecordPtr tabPtr;
+ FragmentstorePtr fragPtr;
+
+ for(tabPtr.i = 0; tabPtr.i < ctabFileSize; tabPtr.i++)
+ {
+ ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
+ if (tabPtr.p->tabStatus != TabRecord::TS_ACTIVE)
+ continue;
+
+ for(Uint32 fid = 0; fid<tabPtr.p->totalfragments; fid++)
+ {
+ getFragstore(tabPtr.p, fid, fragPtr);
+ ndbout_c("tab: %d frag: %d gci: %d\n -- storedReplicas:",
+ tabPtr.i, fid, SYSFILE->newestRestorableGCI);
+
+ Uint32 i;
+ ReplicaRecordPtr replicaPtr;
+ replicaPtr.i = fragPtr.p->storedReplicas;
+ for(; replicaPtr.i != RNIL; replicaPtr.i = replicaPtr.p->nextReplica)
+ {
+ ptrCheckGuard(replicaPtr, creplicaFileSize, replicaRecord);
+ ndbout_c(" node: %d initialGci: %d nextLcp: %d noCrashedReplicas: %d",
+ replicaPtr.p->procNode,
+ replicaPtr.p->initialGci,
+ replicaPtr.p->nextLcp,
+ replicaPtr.p->noCrashedReplicas);
+ for(i = 0; i<MAX_LCP_STORED; i++)
+ {
+ ndbout_c(" i: %d %s : lcpId: %d maxGci Completed: %d Started: %d",
+ i,
+ (replicaPtr.p->lcpStatus[i] == ZVALID ?"VALID":"INVALID"),
+ replicaPtr.p->lcpId[i],
+ replicaPtr.p->maxGciCompleted[i],
+ replicaPtr.p->maxGciStarted[i]);
+ }
+
+ for (i = 0; i < 8; i++)
+ {
+ ndbout_c(" crashed replica: %d replicaLastGci: %d createGci: %d",
+ i,
+ replicaPtr.p->replicaLastGci[i],
+ replicaPtr.p->createGci[i]);
+ }
+ }
+ ndbout_c(" -- oldStoredReplicas");
+ replicaPtr.i = fragPtr.p->oldStoredReplicas;
+ for(; replicaPtr.i != RNIL; replicaPtr.i = replicaPtr.p->nextReplica)
+ {
+ ptrCheckGuard(replicaPtr, creplicaFileSize, replicaRecord);
+ for(i = 0; i<MAX_LCP_STORED; i++)
+ {
+ ndbout_c(" i: %d %s : lcpId: %d maxGci Completed: %d Started: %d",
+ i,
+ (replicaPtr.p->lcpStatus[i] == ZVALID ?"VALID":"INVALID"),
+ replicaPtr.p->lcpId[i],
+ replicaPtr.p->maxGciCompleted[i],
+ replicaPtr.p->maxGciStarted[i]);
+ }
+
+ for (i = 0; i < 8; i++)
+ {
+ ndbout_c(" crashed replica: %d replicaLastGci: %d createGci: %d",
+ i,
+ replicaPtr.p->replicaLastGci[i],
+ replicaPtr.p->createGci[i]);
+ }
+ }
+ }
+ }
+}
+
void Dbdih::startFragment(Signal* signal, Uint32 tableId, Uint32 fragId)
{
Uint32 TloopCount = 0;
@@ -8857,6 +9213,7 @@
/* SEARCH FOR STORED REPLICAS THAT CAN BE USED TO RESTART THE SYSTEM. */
/* ----------------------------------------------------------------------- */
searchStoredReplicas(fragPtr);
+
if (cnoOfCreateReplicas == 0) {
/* --------------------------------------------------------------------- */
/* THERE WERE NO STORED REPLICAS AVAILABLE THAT CAN SERVE AS REPLICA TO*/
@@ -8869,6 +9226,10 @@
char buf[64];
BaseString::snprintf(buf, sizeof(buf), "table: %d fragment: %d gci: %d",
tableId, fragId, SYSFILE->newestRestorableGCI);
+
+ ndbout_c(buf);
+ dump_replica_info();
+
progError(__LINE__, NDBD_EXIT_NO_RESTORABLE_REPLICA, buf);
ndbrequire(false);
return;
@@ -8945,8 +9306,8 @@
// otherwise we have a problem.
/* --------------------------------------------------------------------- */
jam();
- ndbrequire(senderNodeId == c_nodeStartMaster.startNode);
- nodeRestartStartRecConfLab(signal);
+ ndbout_c("startNextCopyFragment");
+ startNextCopyFragment(signal, findTakeOver(senderNodeId));
return;
} else {
/* --------------------------------------------------------------------- */
@@ -9975,9 +10336,11 @@
}
void Dbdih::findReplica(ReplicaRecordPtr& replicaPtr,
- Fragmentstore* fragPtrP, Uint32 nodeId)
+ Fragmentstore* fragPtrP,
+ Uint32 nodeId,
+ bool old)
{
- replicaPtr.i = fragPtrP->storedReplicas;
+ replicaPtr.i = old ? fragPtrP->oldStoredReplicas : fragPtrP->storedReplicas;
while(replicaPtr.i != RNIL){
ptrCheckGuard(replicaPtr, creplicaFileSize, replicaRecord);
if (replicaPtr.p->procNode == nodeId) {
@@ -11234,6 +11597,7 @@
cnoHotSpare = 0;
cnoOfActiveTables = 0;
cnoOfNodeGroups = 0;
+ c_nextNodeGroup = 0;
cnoReplicas = 0;
coldgcp = 0;
coldGcpId = 0;
@@ -11253,6 +11617,7 @@
c_newest_restorable_gci = 0;
cverifyQueueCounter = 0;
cwaitLcpSr = false;
+ c_nextLogPart = 0;
nodeResetStart();
c_nodeStartMaster.wait = ZFALSE;
@@ -11260,7 +11625,7 @@
memset(&sysfileData[0], 0, sizeof(sysfileData));
const ndb_mgm_configuration_iterator * p =
- theConfiguration.getOwnConfigIterator();
+ m_ctx.m_config.getOwnConfigIterator();
ndbrequire(p != 0);
c_lcpState.clcpDelay = 20;
@@ -11339,6 +11704,8 @@
SYSFILE->takeOver[i] = 0;
}//for
Sysfile::setInitialStartOngoing(SYSFILE->systemRestartBits);
+ srand(time(0));
+ globalData.m_restart_seq = SYSFILE->m_restart_seq = 0;
}//Dbdih::initRestartInfo()
/*--------------------------------------------------------------------*/
@@ -12089,6 +12456,8 @@
jam();
fragPtr.p->distributionKey = TdistKey;
}//if
+
+ fragPtr.p->m_log_part_id = readPageWord(rf);
}//Dbdih::readFragment()
Uint32 Dbdih::readPageWord(RWFragment* rf)
@@ -13182,6 +13551,7 @@
writePageWord(wf, fragPtr.p->noStoredReplicas);
writePageWord(wf, fragPtr.p->noOldStoredReplicas);
writePageWord(wf, fragPtr.p->distributionKey);
+ writePageWord(wf, fragPtr.p->m_log_part_id);
}//Dbdih::writeFragment()
void Dbdih::writePageWord(RWFragment* wf, Uint32 dataWord)
@@ -13248,7 +13618,7 @@
signal->theData[0] = filePtr.p->fileRef;
signal->theData[1] = reference();
signal->theData[2] = filePtr.i;
- signal->theData[3] = ZLIST_OF_PAIRS;
+ signal->theData[3] = ZLIST_OF_PAIRS_SYNCH;
signal->theData[4] = ZVAR_NO_WORD;
signal->theData[5] = tab->noPages;
for (Uint32 i = 0; i < tab->noPages; i++) {
@@ -13611,7 +13981,7 @@
if (signal->getLength() == 1)
{
const ndb_mgm_configuration_iterator * p =
- theConfiguration.getOwnConfigIterator();
+ m_ctx.m_config.getOwnConfigIterator();
ndbrequire(p != 0);
ndb_mgm_get_int_parameter(p, CFG_DB_GCP_INTERVAL, &cgcpDelay);
}
@@ -14736,13 +15106,14 @@
{
Uint32 masterVersion = getNodeInfo(cmasterNodeId).m_version;
- unsigned int get_major = getMajor(masterVersion);
- unsigned int get_minor = getMinor(masterVersion);
- unsigned int get_build = getBuild(masterVersion);
-
- ndbrequire(get_major == 4 || get_major == 5);
+ const unsigned int get_major = getMajor(masterVersion);
+ const unsigned int get_minor = getMinor(masterVersion);
+ const unsigned int get_build = getBuild(masterVersion);
+ ndbrequire(get_major >= 4);
if (masterVersion < NDBD_DICT_LOCK_VERSION_5 ||
+ masterVersion < NDBD_DICT_LOCK_VERSION_5_1 &&
+ get_major == 5 && get_minor == 1 ||
ERROR_INSERTED(7176)) {
jam();
@@ -14813,10 +15184,13 @@
{
Uint32 masterVersion = getNodeInfo(cmasterNodeId).m_version;
- unsigned int get_major = getMajor(masterVersion);
- ndbrequire(get_major == 4 || get_major == 5);
+ const unsigned int get_major = getMajor(masterVersion);
+ const unsigned int get_minor = getMinor(masterVersion);
+ ndbrequire(get_major >= 4);
if (masterVersion < NDBD_DICT_LOCK_VERSION_5 ||
+ masterVersion < NDBD_DICT_LOCK_VERSION_5_1 &&
+ get_major == 5 && get_minor == 1 ||
ERROR_INSERTED(7176)) {
return;
}
--- 1.43.9.1/ndb/src/ndbapi/NdbTransaction.cpp 2006-09-26 13:58:38 +02:00
+++ 1.58/storage/ndb/src/ndbapi/NdbTransaction.cpp 2006-09-26 13:58:38 +02:00
@@ -32,6 +32,8 @@
#include <signaldata/TcKeyFailConf.hpp>
#include <signaldata/TcHbRep.hpp>
+Uint64 g_latest_trans_gci = 0;
+
/*****************************************************************************
NdbTransaction( Ndb* aNdb );
@@ -434,7 +436,7 @@
//------------------------------------------------------------------------
Ndb* tNdb = theNdb;
- Uint32 timeout = TransporterFacade::instance()->m_waitfor_timeout;
+ Uint32 timeout = theNdb->theImpl->m_transporter_facade->m_waitfor_timeout;
m_waitForReply = false;
executeAsynchPrepare(aTypeOfExec, NULL, NULL, abortOption);
if (m_waitForReply){
@@ -706,7 +708,7 @@
tcHbRep->transId1 = tTransId1;
tcHbRep->transId2 = tTransId2;
- TransporterFacade *tp = TransporterFacade::instance();
+ TransporterFacade *tp = theNdb->theImpl->m_transporter_facade;
tp->lock_mutex();
const int res = tp->sendSignal(tSignal,theDBnode);
tp->unlock_mutex();
@@ -812,7 +814,7 @@
*************************************************************************/
NdbApiSignal tSignal(tNdb->theMyRef);
Uint32 tTransId1, tTransId2;
- TransporterFacade *tp = TransporterFacade::instance();
+ TransporterFacade *tp = theNdb->theImpl->m_transporter_facade;
int tReturnCode;
tTransId1 = (Uint32) theTransactionId;
@@ -859,7 +861,7 @@
{
NdbApiSignal tSignal(theNdb->theMyRef);
Uint32 tTransId1, tTransId2;
- TransporterFacade *tp = TransporterFacade::instance();
+ TransporterFacade *tp = theNdb->theImpl->m_transporter_facade;
int tReturnCode;
tTransId1 = (Uint32) theTransactionId;
@@ -970,12 +972,12 @@
}//NdbTransaction::releaseScanOperations()
/*****************************************************************************
-void releaseExecutedScanOperation();
+void releaseScanOperation();
Remark: Release scan op when hupp'ed trans closed (save memory)
******************************************************************************/
void
-NdbTransaction::releaseExecutedScanOperation(NdbIndexScanOperation* cursorOp)
+NdbTransaction::releaseScanOperation(NdbIndexScanOperation* cursorOp)
{
DBUG_ENTER("NdbTransaction::releaseExecutedScanOperation");
DBUG_PRINT("enter", ("this=0x%x op=0x%x", (UintPtr)this, (UintPtr)cursorOp));
@@ -1201,9 +1203,9 @@
if(tOp)
{
tOp->m_currentTable = table;
+ // Mark that this really an NdbIndexScanOperation
+ tOp->m_type = NdbOperation::OrderedIndexScan;
}
- // Mark that this really an NdbIndexScanOperation
- tOp->m_type = NdbOperation::OrderedIndexScan;
return tOp;
} else {
setOperationErrorCodeAbort(4271);
@@ -1577,6 +1579,9 @@
theCommitStatus = Committed;
theCompletionStatus = CompletedSuccess;
theGlobalCheckpointId = commitConf->gci;
+ // theGlobalCheckpointId == 0 if NoOp transaction
+ if (theGlobalCheckpointId)
+ g_latest_trans_gci = theGlobalCheckpointId;
return 0;
} else {
#ifdef NDB_NO_DROPPED_SIGNAL
@@ -1755,6 +1760,8 @@
if (tCommitFlag == 1) {
theCommitStatus = Committed;
theGlobalCheckpointId = tGCI;
+ assert(tGCI);
+ g_latest_trans_gci = tGCI;
} else if ((tNoComp >= tNoSent) &&
(theLastExecOpInList->theCommitIndicator == 1)){
@@ -1931,6 +1938,8 @@
if (tCommitFlag == 1) {
theCommitStatus = Committed;
theGlobalCheckpointId = tGCI;
+ assert(tGCI);
+ g_latest_trans_gci = tGCI;
} else if ((tNoComp >= tNoSent) &&
(theLastExecOpInList->theCommitIndicator == 1)){
/**********************************************************************/
--- 1.66.13.1/ndb/src/ndbapi/NdbScanOperation.cpp 2006-09-26 13:58:38 +02:00
+++ 1.94/storage/ndb/src/ndbapi/NdbScanOperation.cpp 2006-09-26 13:58:38 +02:00
@@ -50,6 +50,7 @@
m_receivers = 0;
m_array = new Uint32[1]; // skip if on delete in fix_receivers
theSCAN_TABREQ = 0;
+ m_executed = false;
}
NdbScanOperation::~NdbScanOperation()
@@ -111,6 +112,7 @@
theNdbCon->theMagicNumber = 0xFE11DF;
theNoOfTupKeyLeft = tab->m_noOfDistributionKeys;
m_read_range_no = 0;
+ m_executed = false;
return 0;
}
@@ -162,6 +164,16 @@
}
m_keyInfo = ((scan_flags & SF_KeyInfo) || lockExcl) ? 1 : 0;
+ bool tupScan = (scan_flags & SF_TupScan);
+
+#if 1 // XXX temp for testing
+ { char* p = getenv("NDB_USE_TUPSCAN");
+ if (p != 0) {
+ unsigned n = atoi(p); // 0-10
+ if ((unsigned int) (::time(0) % 10) < n) tupScan = true;
+ }
+ }
+#endif
bool rangeScan = false;
if (m_accessTable->m_indexType == NdbDictionary::Index::OrderedIndex)
@@ -177,12 +189,9 @@
theStatus = GetValue;
theOperationType = OpenRangeScanRequest;
rangeScan = true;
- }
-
- bool tupScan = (scan_flags & SF_TupScan);
- if (tupScan && rangeScan)
tupScan = false;
-
+ }
+
if (rangeScan && (scan_flags & SF_OrderBy))
parallel = fragCount;
@@ -202,7 +211,7 @@
theSCAN_TABREQ->setSignal(GSN_SCAN_TABREQ);
ScanTabReq * req = CAST_PTR(ScanTabReq, theSCAN_TABREQ->getDataPtrSend());
req->apiConnectPtr = theNdbCon->theTCConPtr;
- req->tableId = m_accessTable->m_tableId;
+ req->tableId = m_accessTable->m_id;
req->tableSchemaVersion = m_accessTable->m_version;
req->storedProcId = 0xFFFF;
req->buddyConPtr = theNdbCon->theBuddyConPtr;
@@ -363,7 +372,7 @@
int
NdbScanOperation::executeCursor(int nodeId){
NdbTransaction * tCon = theNdbCon;
- TransporterFacade* tp = TransporterFacade::instance();
+ TransporterFacade* tp = theNdb->theImpl->m_transporter_facade;
Guard guard(tp->theMutexPtr);
Uint32 magic = tCon->theMagicNumber;
@@ -385,6 +394,7 @@
if (doSendScan(nodeId) == -1)
return -1;
+ m_executed= true; // Mark operation as executed
return 0;
} else {
if (!(tp->get_node_stopping(nodeId) &&
@@ -464,18 +474,28 @@
}
Uint32 nodeId = theNdbCon->theDBnode;
- TransporterFacade* tp = TransporterFacade::instance();
- Guard guard(tp->theMutexPtr);
- if(theError.code)
- return -1;
+ TransporterFacade* tp = theNdb->theImpl->m_transporter_facade;
+ /*
+ The PollGuard has an implicit call of unlock_and_signal through the
+ ~PollGuard method. This method is called implicitly by the compiler
+ in all places where the object is out of context due to a return,
+ break, continue or simply end of statement block
+ */
+ PollGuard poll_guard(tp, &theNdb->theImpl->theWaiter,
+ theNdb->theNdbBlockNumber);
- Uint32 seq = theNdbCon->theNodeSequence;
- if(seq == tp->getNodeSequence(nodeId) && send_next_scan(idx, false,
- forceSend) == 0){
+ const Uint32 seq = theNdbCon->theNodeSequence;
+
+ if(theError.code)
+ {
+ goto err4;
+ }
+
+ if(seq == tp->getNodeSequence(nodeId) && send_next_scan(idx, false) == 0)
+ {
idx = m_current_api_receiver;
last = m_api_receivers_count;
-
Uint32 timeout = tp->m_waitfor_timeout;
do {
@@ -502,10 +522,8 @@
/**
* No completed...
*/
- theNdb->theImpl->theWaiter.m_node = nodeId;
- theNdb->theImpl->theWaiter.m_state = WAIT_SCAN;
- int return_code = theNdb->receiveResponse(3*timeout);
- if (return_code == 0 && seq == tp->getNodeSequence(nodeId)) {
+ int ret_code= poll_guard.wait_scan(3*timeout, nodeId, forceSend);
+ if (ret_code == 0 && seq == tp->getNodeSequence(nodeId)) {
continue;
} else {
idx = last;
@@ -555,6 +573,10 @@
if(theError.code == 0)
setErrorCode(4028); // seq changed = Node fail
break;
+ case -4:
+err4:
+ setErrorCode(theError.code);
+ break;
}
theNdbCon->theTransactionIsStarted = false;
@@ -564,8 +586,8 @@
}
int
-NdbScanOperation::send_next_scan(Uint32 cnt, bool stopScanFlag,
- bool forceSend){
+NdbScanOperation::send_next_scan(Uint32 cnt, bool stopScanFlag)
+{
if(cnt > 0){
NdbApiSignal tSignal(theNdb->theMyRef);
tSignal.setSignal(GSN_SCAN_NEXTREQ);
@@ -600,7 +622,7 @@
if(sent)
{
Uint32 nodeId = theNdbCon->theDBnode;
- TransporterFacade * tp = TransporterFacade::instance();
+ TransporterFacade * tp = theNdb->theImpl->m_transporter_facade;
if(cnt > 21){
tSignal.setLength(4);
LinearSectionPtr ptr[3];
@@ -612,9 +634,6 @@
ret = tp->sendSignal(&tSignal, nodeId);
}
}
-
- if (!ret) checkForceSend(forceSend);
-
m_sent_receivers_count = last + sent;
m_api_receivers_count -= cnt;
m_current_api_receiver = 0;
@@ -624,15 +643,6 @@
return 0;
}
-void NdbScanOperation::checkForceSend(bool forceSend)
-{
- if (forceSend) {
- TransporterFacade::instance()->forceSend(theNdb->theNdbBlockNumber);
- } else {
- TransporterFacade::instance()->checkForceSend(theNdb->theNdbBlockNumber);
- }//if
-}
-
int
NdbScanOperation::prepareSend(Uint32 TC_ConnectPtr, Uint64 TransactionId)
{
@@ -667,10 +677,16 @@
m_conf_receivers_count,
m_sent_receivers_count);
- TransporterFacade* tp = TransporterFacade::instance();
- Guard guard(tp->theMutexPtr);
- close_impl(tp, forceSend);
-
+ TransporterFacade* tp = theNdb->theImpl->m_transporter_facade;
+ /*
+ The PollGuard has an implicit call of unlock_and_signal through the
+ ~PollGuard method. This method is called implicitly by the compiler
+ in all places where the object is out of context due to a return,
+ break, continue or simply end of statement block
+ */
+ PollGuard poll_guard(tp, &theNdb->theImpl->theWaiter,
+ theNdb->theNdbBlockNumber);
+ close_impl(tp, forceSend, &poll_guard);
}
NdbConnection* tCon = theNdbCon;
@@ -793,6 +809,7 @@
*/
Uint32 reqInfo = req->requestInfo;
ScanTabReq::setKeyinfoFlag(reqInfo, keyInfo);
+ ScanTabReq::setNoDiskFlag(reqInfo, m_no_disk_flag);
req->requestInfo = reqInfo;
for(Uint32 i = 0; i<theParallelism; i++){
@@ -843,7 +860,7 @@
req->requestInfo = tmp;
tSignal->setLength(ScanTabReq::StaticLength + theDistrKeyIndicator_);
- TransporterFacade *tp = TransporterFacade::instance();
+ TransporterFacade *tp = theNdb->theImpl->m_transporter_facade;
LinearSectionPtr ptr[3];
ptr[0].p = m_prepared_receivers;
ptr[0].sz = theParallelism;
@@ -927,13 +944,20 @@
* the scan process.
****************************************************************************/
int
-NdbScanOperation::getKeyFromKEYINFO20(Uint32* data, unsigned size)
+NdbScanOperation::getKeyFromKEYINFO20(Uint32* data, Uint32 & size)
{
NdbRecAttr * tRecAttr = m_curr_row;
if(tRecAttr)
{
const Uint32 * src = (Uint32*)tRecAttr->aRef();
- memcpy(data, src, 4*size);
+
+ assert(tRecAttr->get_size_in_bytes() > 0);
+ assert(tRecAttr->get_size_in_bytes() < 65536);
+ const Uint32 len = (tRecAttr->get_size_in_bytes() + 3)/4-1;
+
+ assert(size >= len);
+ memcpy(data, src, 4*len);
+ size = len;
return 0;
}
return -1;
@@ -958,8 +982,10 @@
}
pTrans->theSimpleState = 0;
- const Uint32 len = (tRecAttr->attrSize() * tRecAttr->arraySize() + 3)/4-1;
-
+ assert(tRecAttr->get_size_in_bytes() > 0);
+ assert(tRecAttr->get_size_in_bytes() < 65536);
+ const Uint32 len = (tRecAttr->get_size_in_bytes() + 3)/4-1;
+
newOp->theTupKeyLen = len;
newOp->theOperationType = opType;
switch (opType) {
@@ -1057,23 +1083,23 @@
int
NdbIndexScanOperation::setBound(const char* anAttrName, int type,
- const void* aValue, Uint32 len)
+ const void* aValue)
{
- return setBound(m_accessTable->getColumn(anAttrName), type, aValue, len);
+ return setBound(m_accessTable->getColumn(anAttrName), type, aValue);
}
int
NdbIndexScanOperation::setBound(Uint32 anAttrId, int type,
- const void* aValue, Uint32 len)
+ const void* aValue)
{
- return setBound(m_accessTable->getColumn(anAttrId), type, aValue, len);
+ return setBound(m_accessTable->getColumn(anAttrId), type, aValue);
}
int
NdbIndexScanOperation::equal_impl(const NdbColumnImpl* anAttrObject,
- const char* aValue,
- Uint32 len){
- return setBound(anAttrObject, BoundEQ, aValue, len);
+ const char* aValue)
+{
+ return setBound(anAttrObject, BoundEQ, aValue);
}
NdbRecAttr*
@@ -1083,7 +1109,7 @@
return NdbScanOperation::getValue_impl(attrInfo, aValue);
}
- int id = attrInfo->m_attrId; // In "real" table
+ int id = attrInfo->getColumnNo(); // In "real" table
assert(m_accessTable->m_index);
int sz = (int)m_accessTable->m_index->m_key_ids.size();
if(id >= sz || (id = m_accessTable->m_index->m_key_ids[id]) == -1){
@@ -1120,7 +1146,7 @@
*/
int
NdbIndexScanOperation::setBound(const NdbColumnImpl* tAttrInfo,
- int type, const void* aValue, Uint32 len)
+ int type, const void* aValue)
{
if (!tAttrInfo)
{
@@ -1128,24 +1154,23 @@
return -1;
}
if (theOperationType == OpenRangeScanRequest &&
- (0 <= type && type <= 4) &&
- len <= 8000) {
+ (0 <= type && type <= 4)) {
// insert bound type
Uint32 currLen = theTotalNrOfKeyWordInSignal;
Uint32 remaining = KeyInfo::DataLength - currLen;
- Uint32 sizeInBytes = tAttrInfo->m_attrSize * tAttrInfo->m_arraySize;
bool tDistrKey = tAttrInfo->m_distributionKey;
- len = aValue != NULL ? sizeInBytes : 0;
- if (len != sizeInBytes && (len != 0)) {
- setErrorCodeAbort(4209);
- return -1;
- }
+ Uint32 len = 0;
+ if (aValue != NULL)
+ if (! tAttrInfo->get_var_length(aValue, len)) {
+ setErrorCodeAbort(4209);
+ return -1;
+ }
// insert attribute header
Uint32 tIndexAttrId = tAttrInfo->m_attrId;
Uint32 sizeInWords = (len + 3) / 4;
- AttributeHeader ah(tIndexAttrId, sizeInWords);
+ AttributeHeader ah(tIndexAttrId, sizeInWords << 2);
const Uint32 ahValue = ah.m_value;
const Uint32 align = (UintPtr(aValue) & 7);
@@ -1239,6 +1264,31 @@
return -1;
}
+Uint32
+NdbIndexScanOperation::getKeyFromSCANTABREQ(Uint32* data, Uint32 size)
+{
+ DBUG_ENTER("NdbIndexScanOperation::getKeyFromSCANTABREQ");
+ assert(size >= theTotalNrOfKeyWordInSignal);
+ size = theTotalNrOfKeyWordInSignal;
+ NdbApiSignal* tSignal = theSCAN_TABREQ->next();
+ Uint32 pos = 0;
+ while (pos < size) {
+ assert(tSignal != NULL);
+ Uint32* tData = tSignal->getDataPtrSend();
+ Uint32 rem = size - pos;
+ if (rem > KeyInfo::DataLength)
+ rem = KeyInfo::DataLength;
+ Uint32 i = 0;
+ while (i < rem) {
+ data[pos + i] = tData[KeyInfo::HeaderLength + i];
+ i++;
+ }
+ pos += rem;
+ }
+ DBUG_DUMP("key", (char*)data, size << 2);
+ DBUG_RETURN(size);
+}
+
int
NdbIndexScanOperation::readTuples(LockMode lm,
Uint32 scan_flags,
@@ -1338,10 +1388,11 @@
return (r1_null ? -1 : 1) * jdir;
}
const NdbColumnImpl & col = NdbColumnImpl::getImpl(* r1->m_column);
- Uint32 len = r1->theAttrSize * r1->theArraySize;
+ Uint32 len1 = r1->get_size_in_bytes();
+ Uint32 len2 = r2->get_size_in_bytes();
if(!r1_null){
const NdbSqlUtil::Type& sqlType = NdbSqlUtil::getType(col.m_type);
- int r = (*sqlType.m_cmp)(col.m_cs, d1, len, d2, len, true);
+ int r = (*sqlType.m_cmp)(col.m_cs, d1, len1, d2, len2, true);
if(r){
assert(r != NdbSqlUtil::CmpUnknown);
return r * jdir;
@@ -1379,22 +1430,27 @@
if(fetchNeeded){
if(fetchAllowed){
if(DEBUG_NEXT_RESULT) ndbout_c("performing fetch...");
- TransporterFacade* tp = TransporterFacade::instance();
- Guard guard(tp->theMutexPtr);
+ TransporterFacade* tp = theNdb->theImpl->m_transporter_facade;
+ /*
+ The PollGuard has an implicit call of unlock_and_signal through the
+ ~PollGuard method. This method is called implicitly by the compiler
+ in all places where the object is out of context due to a return,
+ break, continue or simply end of statement block
+ */
+ PollGuard poll_guard(tp, &theNdb->theImpl->theWaiter,
+ theNdb->theNdbBlockNumber);
if(theError.code)
return -1;
Uint32 seq = theNdbCon->theNodeSequence;
Uint32 nodeId = theNdbCon->theDBnode;
Uint32 timeout = tp->m_waitfor_timeout;
if(seq == tp->getNodeSequence(nodeId) &&
- !send_next_scan_ordered(s_idx, forceSend)){
+ !send_next_scan_ordered(s_idx)){
Uint32 tmp = m_sent_receivers_count;
s_idx = m_current_api_receiver;
while(m_sent_receivers_count > 0 && !theError.code){
- theNdb->theImpl->theWaiter.m_node = nodeId;
- theNdb->theImpl->theWaiter.m_state = WAIT_SCAN;
- int return_code = theNdb->receiveResponse(3*timeout);
- if (return_code == 0 && seq == tp->getNodeSequence(nodeId)) {
+ int ret_code= poll_guard.wait_scan(3*timeout, nodeId, forceSend);
+ if (ret_code == 0 && seq == tp->getNodeSequence(nodeId)) {
continue;
}
if(DEBUG_NEXT_RESULT) ndbout_c("return -1");
@@ -1481,7 +1537,8 @@
}
int
-NdbIndexScanOperation::send_next_scan_ordered(Uint32 idx, bool forceSend){
+NdbIndexScanOperation::send_next_scan_ordered(Uint32 idx)
+{
if(idx == theParallelism)
return 0;
@@ -1516,15 +1573,16 @@
m_sent_receivers_count = last + 1;
Uint32 nodeId = theNdbCon->theDBnode;
- TransporterFacade * tp = TransporterFacade::instance();
+ TransporterFacade * tp = theNdb->theImpl->m_transporter_facade;
tSignal.setLength(4+1);
int ret= tp->sendSignal(&tSignal, nodeId);
- if (!ret) checkForceSend(forceSend);
return ret;
}
int
-NdbScanOperation::close_impl(TransporterFacade* tp, bool forceSend){
+NdbScanOperation::close_impl(TransporterFacade* tp, bool forceSend,
+ PollGuard *poll_guard)
+{
Uint32 seq = theNdbCon->theNodeSequence;
Uint32 nodeId = theNdbCon->theDBnode;
@@ -1535,15 +1593,12 @@
}
Uint32 timeout = tp->m_waitfor_timeout;
-
/**
* Wait for outstanding
*/
while(theError.code == 0 && m_sent_receivers_count)
{
- theNdb->theImpl->theWaiter.m_node = nodeId;
- theNdb->theImpl->theWaiter.m_state = WAIT_SCAN;
- int return_code = theNdb->receiveResponse(3*timeout);
+ int return_code= poll_guard->wait_scan(3*timeout, nodeId, forceSend);
switch(return_code){
case 0:
break;
@@ -1600,7 +1655,7 @@
}
// Send close scan
- if(send_next_scan(api+conf, true, forceSend) == -1)
+ if(send_next_scan(api+conf, true) == -1)
{
theNdbCon->theReleaseOnClose = true;
return -1;
@@ -1611,9 +1666,7 @@
*/
while(m_sent_receivers_count+m_api_receivers_count+m_conf_receivers_count)
{
- theNdb->theImpl->theWaiter.m_node = nodeId;
- theNdb->theImpl->theWaiter.m_state = WAIT_SCAN;
- int return_code = theNdb->receiveResponse(3*timeout);
+ int return_code= poll_guard->wait_scan(3*timeout, nodeId, forceSend);
switch(return_code){
case 0:
break;
@@ -1652,13 +1705,20 @@
NdbScanOperation::restart(bool forceSend)
{
- TransporterFacade* tp = TransporterFacade::instance();
- Guard guard(tp->theMutexPtr);
+ TransporterFacade* tp = theNdb->theImpl->m_transporter_facade;
+ /*
+ The PollGuard has an implicit call of unlock_and_signal through the
+ ~PollGuard method. This method is called implicitly by the compiler
+ in all places where the object is out of context due to a return,
+ break, continue or simply end of statement block
+ */
+ PollGuard poll_guard(tp, &theNdb->theImpl->theWaiter,
+ theNdb->theNdbBlockNumber);
Uint32 nodeId = theNdbCon->theDBnode;
{
int res;
- if((res= close_impl(tp, forceSend)))
+ if((res= close_impl(tp, forceSend, &poll_guard)))
{
return res;
}
@@ -1672,7 +1732,6 @@
theError.code = 0;
if (doSendScan(nodeId) == -1)
return -1;
-
return 0;
}
@@ -1681,9 +1740,16 @@
int res;
{
- TransporterFacade* tp = TransporterFacade::instance();
- Guard guard(tp->theMutexPtr);
- res= close_impl(tp, forceSend);
+ TransporterFacade* tp = theNdb->theImpl->m_transporter_facade;
+ /*
+ The PollGuard has an implicit call of unlock_and_signal through the
+ ~PollGuard method. This method is called implicitly by the compiler
+ in all places where the object is out of context due to a return,
+ break, continue or simply end of statement block
+ */
+ PollGuard poll_guard(tp, &theNdb->theImpl->theWaiter,
+ theNdb->theNdbBlockNumber);
+ res= close_impl(tp, forceSend, &poll_guard);
}
if(!res)
| Thread |
|---|
| • bk commit into 5.1 tree (jonas:1.2296) | jonas | 26 Sep |
