MySQL数据库

开发平台：

Visual C++

DbaccMain.cpp：源码内容

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 */
#define DBACC_C
#include "Dbacc.hpp"
#include <AttributeHeader.hpp>
#include <signaldata/AccFrag.hpp>
#include <signaldata/AccScan.hpp>
#include <signaldata/AccLock.hpp>
#include <signaldata/EventReport.hpp>
#include <signaldata/FsConf.hpp>
#include <signaldata/FsRef.hpp>
#include <signaldata/FsRemoveReq.hpp>
#include <signaldata/DropTab.hpp>
#include <signaldata/DumpStateOrd.hpp>
// TO_DO_RONM is a label for comments on what needs to be improved in future versions
// when more time is given.
#ifdef VM_TRACE
#define DEBUG(x) ndbout << "DBACC: "<< x << endl;
#else
#define DEBUG(x)
#endif
Uint32
Dbacc::remainingUndoPages(){
Uint32 HeadPage = cundoposition >> ZUNDOPAGEINDEXBITS;
Uint32 TailPage = clastUndoPageIdWritten;
// Head must be larger or same as tail
ndbrequire(HeadPage>=TailPage);
Uint32 UsedPages = HeadPage - TailPage;
Int32 Remaining = cundopagesize - UsedPages;
// There can not be more than cundopagesize remaining
if (Remaining <= 0){
// No more undolog, crash node
progError(__LINE__,
ERR_NO_MORE_UNDOLOG,
"There are more than 1Mbyte undolog writes outstanding");
}
return Remaining;
}
void
Dbacc::updateLastUndoPageIdWritten(Signal* signal, Uint32 aNewValue){
if (remainingUndoPages() < ZMIN_UNDO_PAGES_AT_COMMIT) {
clastUndoPageIdWritten = aNewValue;
if (remainingUndoPages() >= ZMIN_UNDO_PAGES_AT_COMMIT) {
jam();
EXECUTE_DIRECT(DBLQH, GSN_ACC_COM_UNBLOCK, signal, 1);
jamEntry();
}//if
} else {
clastUndoPageIdWritten = aNewValue;
}//if
}//Dbacc::updateLastUndoPageIdWritten()
void
Dbacc::updateUndoPositionPage(Signal* signal, Uint32 aNewValue){
if (remainingUndoPages() >= ZMIN_UNDO_PAGES_AT_COMMIT) {
cundoposition = aNewValue;
if (remainingUndoPages() < ZMIN_UNDO_PAGES_AT_COMMIT) {
jam();
EXECUTE_DIRECT(DBLQH, GSN_ACC_COM_BLOCK, signal, 1);
jamEntry();
}//if
} else {
cundoposition = aNewValue;
}//if
}//Dbacc::updateUndoPositionPage()
// Signal entries and statement blocks
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* */
/* COMMON SIGNAL RECEPTION MODULE */
/* */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* ******************--------------------------------------------------------------- */
/* CONTINUEB CONTINUE SIGNAL */
/* ******************------------------------------+ */
/* SENDER: ACC, LEVEL B */
void Dbacc::execCONTINUEB(Signal* signal)
{
Uint32 tcase;
jamEntry();
tcase = signal->theData[0];
tdata0 = signal->theData[1];
tresult = 0;
switch (tcase) {
case ZLOAD_BAL_LCP_TIMER:
if (clblPageOver == 0) {
jam();
clblPageCounter = clblPagesPerTick;
} else {
if (clblPageOver > clblPagesPerTick) {
jam();
clblPageOver = clblPageOver - clblPagesPerTick;
} else {
jam();
clblPageOver = 0;
clblPageCounter = clblPagesPerTick - clblPageOver;
}//if
}//if
signal->theData[0] = ZLOAD_BAL_LCP_TIMER;
sendSignalWithDelay(cownBlockref, GSN_CONTINUEB, signal, 100, 1);
return;
break;
case ZINITIALISE_RECORDS:
jam();
initialiseRecordsLab(signal, signal->theData[3], signal->theData[4]);
return;
break;
case ZSR_READ_PAGES_ALLOC:
jam();
fragrecptr.i = tdata0;
ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
srReadPagesAllocLab(signal);
return;
break;
case ZSTART_UNDO:
jam();
startUndoLab(signal);
return;
break;
case ZSEND_SCAN_HBREP:
jam();
sendScanHbRep(signal, tdata0);
break;
case ZREL_ROOT_FRAG:
{
jam();
Uint32 tableId = signal->theData[1];
releaseRootFragResources(signal, tableId);
break;
}
case ZREL_FRAG:
{
jam();
Uint32 fragIndex = signal->theData[1];
releaseFragResources(signal, fragIndex);
break;
}
case ZREL_DIR:
{
jam();
Uint32 fragIndex = signal->theData[1];
Uint32 dirIndex = signal->theData[2];
Uint32 startIndex = signal->theData[3];
releaseDirResources(signal, fragIndex, dirIndex, startIndex);
break;
}
case ZREPORT_MEMORY_USAGE:{
jam();
static int c_currentMemUsed = 0;
int now = (cnoOfAllocatedPages * 100)/cpagesize;
const int thresholds[] = { 99, 90, 80, 0};
Uint32 i = 0;
const Uint32 sz = sizeof(thresholds)/sizeof(thresholds[0]);
for(i = 0; i<sz; i++){
if(now >= thresholds[i]){
now = thresholds[i];
break;
}
}
if(now != c_currentMemUsed){
reportMemoryUsage(signal, now > c_currentMemUsed ? 1 : -1);
}
c_currentMemUsed = now;
signal->theData[0] = ZREPORT_MEMORY_USAGE;
sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, 2000, 1);
return;
}
case ZLCP_OP_WRITE_RT_BREAK:
{
operationRecPtr.i= signal->theData[1];
fragrecptr.i= signal->theData[2];
lcpConnectptr.i= signal->theData[3];
ptrCheckGuard(operationRecPtr, coprecsize, operationrec);
ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
ptrCheckGuard(lcpConnectptr, clcpConnectsize, lcpConnectrec);
lcp_write_op_to_undolog(signal);
return;
}
default:
ndbrequire(false);
break;
}//switch
return;
}//Dbacc::execCONTINUEB()
/* ******************--------------------------------------------------------------- */
/* FSCLOSECONF CLOSE FILE CONF */
/* ******************------------------------------+ */
/* SENDER: FS, LEVEL B */
void Dbacc::execFSCLOSECONF(Signal* signal)
{
jamEntry();
fsConnectptr.i = signal->theData[0];
ptrCheckGuard(fsConnectptr, cfsConnectsize, fsConnectrec);
tresult = 0;
switch (fsConnectptr.p->fsState) {
case WAIT_CLOSE_UNDO:
jam();
releaseFsConnRec(signal);
break;
case LCP_CLOSE_DATA:
jam();
checkSyncUndoPagesLab(signal);
return;
break;
case SR_CLOSE_DATA:
jam();
sendaccSrconfLab(signal);
return;
break;
default:
ndbrequire(false);
break;
}//switch
return;
}//Dbacc::execFSCLOSECONF()
/* ******************--------------------------------------------------------------- */
/* FSOPENCONF OPENFILE CONF */
/* ******************------------------------------+ */
/* SENDER: FS, LEVEL B */
void Dbacc::execFSOPENCONF(Signal* signal)
{
jamEntry();
fsConnectptr.i = signal->theData[0];
ptrCheckGuard(fsConnectptr, cfsConnectsize, fsConnectrec);
tuserptr = signal->theData[1];
tresult = 0; /* RESULT CHECK VALUE */
switch (fsConnectptr.p->fsState) {
case WAIT_OPEN_UNDO_LCP:
jam();
lcpOpenUndofileConfLab(signal);
return;
break;
case WAIT_OPEN_UNDO_LCP_NEXT:
jam();
fsConnectptr.p->fsPtr = tuserptr;
return;
break;
case OPEN_UNDO_FILE_SR:
jam();
fsConnectptr.p->fsPtr = tuserptr;
srStartUndoLab(signal);
return;
break;
case WAIT_OPEN_DATA_FILE_FOR_WRITE:
jam();
lcpFsOpenConfLab(signal);
return;
break;
case WAIT_OPEN_DATA_FILE_FOR_READ:
jam();
fsConnectptr.p->fsPtr = tuserptr;
srFsOpenConfLab(signal);
return;
break;
default:
ndbrequire(false);
break;
}//switch
return;
}//Dbacc::execFSOPENCONF()
/* ******************--------------------------------------------------------------- */
/* FSREADCONF OPENFILE CONF */
/* ******************------------------------------+ */
/* SENDER: FS, LEVEL B */
void Dbacc::execFSREADCONF(Signal* signal)
{
jamEntry();
fsConnectptr.i = signal->theData[0];
ptrCheckGuard(fsConnectptr, cfsConnectsize, fsConnectrec);
tresult = 0; /* RESULT CHECK VALUE */
switch (fsConnectptr.p->fsState) {
case WAIT_READ_PAGE_ZERO:
jam();
fragrecptr.i = fsConnectptr.p->fragrecPtr;
ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
srReadPageZeroLab(signal);
return;
break;
case WAIT_READ_DATA:
jam();
fragrecptr.i = fsConnectptr.p->fragrecPtr;
ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
storeDataPageInDirectoryLab(signal);
return;
break;
case READ_UNDO_PAGE:
jam();
srDoUndoLab(signal);
return;
break;
case READ_UNDO_PAGE_AND_CLOSE:
jam();
fsConnectptr.p->fsState = WAIT_CLOSE_UNDO;
/* ************************ */
/* FSCLOSEREQ */
/* ************************ */
signal->theData[0] = fsConnectptr.p->fsPtr;
signal->theData[1] = cownBlockref;
signal->theData[2] = fsConnectptr.i;
signal->theData[3] = 0;
sendSignal(NDBFS_REF, GSN_FSCLOSEREQ, signal, 4, JBA);
/* FLAG = DO NOT DELETE FILE */
srDoUndoLab(signal);
return;
break;
default:
ndbrequire(false);
break;
}//switch
return;
}//Dbacc::execFSREADCONF()
/* ******************--------------------------------------------------------------- */
/* FSWRITECONF OPENFILE CONF */
/* ******************------------------------------+ */
/* SENDER: FS, LEVEL B */
void Dbacc::execFSWRITECONF(Signal* signal)
{
jamEntry();
fsOpptr.i = signal->theData[0];
ptrCheckGuard(fsOpptr, cfsOpsize, fsOprec);
/* FS_OPERATION PTR */
tresult = 0; /* RESULT CHECK VALUE */
fsConnectptr.i = fsOpptr.p->fsConptr;
ptrCheckGuard(fsConnectptr, cfsConnectsize, fsConnectrec);
fragrecptr.i = fsOpptr.p->fsOpfragrecPtr;
ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
switch (fsOpptr.p->fsOpstate) {
case WAIT_WRITE_UNDO:
jam();
updateLastUndoPageIdWritten(signal, fsOpptr.p->fsOpMemPage);
releaseFsOpRec(signal);
if (fragrecptr.p->nrWaitWriteUndoExit == 0) {
jam();
checkSendLcpConfLab(signal);
return;
} else {
jam();
fragrecptr.p->lastUndoIsStored = ZTRUE;
}//if
return;
break;
case WAIT_WRITE_UNDO_EXIT:
jam();
updateLastUndoPageIdWritten(signal, fsOpptr.p->fsOpMemPage);
releaseFsOpRec(signal);
if (fragrecptr.p->nrWaitWriteUndoExit > 0) {
jam();
fragrecptr.p->nrWaitWriteUndoExit--;
}//if
if (fsConnectptr.p->fsState == WAIT_CLOSE_UNDO) {
jam();
/* ************************ */
/* FSCLOSEREQ */
/* ************************ */
signal->theData[0] = fsConnectptr.p->fsPtr;
signal->theData[1] = cownBlockref;
signal->theData[2] = fsConnectptr.i;
signal->theData[3] = ZFALSE;
sendSignal(NDBFS_REF, GSN_FSCLOSEREQ, signal, 4, JBA);
}//if
if (fragrecptr.p->nrWaitWriteUndoExit == 0) {
if (fragrecptr.p->lastUndoIsStored == ZTRUE) {
jam();
fragrecptr.p->lastUndoIsStored = ZFALSE;
checkSendLcpConfLab(signal);
return;
}//if
}//if
return;
break;
case WAIT_WRITE_DATA:
jam();
releaseFsOpRec(signal);
fragrecptr.p->activeDataFilePage += ZWRITEPAGESIZE;
fragrecptr.p->activeDataPage = 0;
rootfragrecptr.i = fragrecptr.p->myroot;
ptrCheckGuard(rootfragrecptr, crootfragmentsize, rootfragmentrec);
lcpConnectptr.i = rootfragrecptr.p->lcpPtr;
ptrCheckGuard(lcpConnectptr, clcpConnectsize, lcpConnectrec);
switch (fragrecptr.p->fragState) {
case LCP_SEND_PAGES:
jam();
savepagesLab(signal);
return;
break;
case LCP_SEND_OVER_PAGES:
jam();
saveOverPagesLab(signal);
return;
break;
case LCP_SEND_ZERO_PAGE:
jam();
saveZeroPageLab(signal);
return;
break;
case WAIT_ZERO_PAGE_STORED:
jam();
lcpCloseDataFileLab(signal);
return;
break;
default:
ndbrequire(false);
return;
break;
}//switch
break;
default:
ndbrequire(false);
break;
}//switch
return;
}//Dbacc::execFSWRITECONF()
/* ------------------------------------------------------------------------- */
/* ------------------------------------------------------------------------- */
/* ------------------------------------------------------------------------- */
/* */
/* END OF COMMON SIGNAL RECEPTION MODULE */
/* */
/* ------------------------------------------------------------------------- */
/* ------------------------------------------------------------------------- */
/* ------------------------------------------------------------------------- */
/* ------------------------------------------------------------------------- */
/* ------------------------------------------------------------------------- */
/* */
/* SYSTEM RESTART MODULE */
/* */
/* ------------------------------------------------------------------------- */
/* ------------------------------------------------------------------------- */
/* ------------------------------------------------------------------------- */
void Dbacc::execNDB_STTOR(Signal* signal)
{
Uint32 tstartphase;
Uint32 tStartType;
jamEntry();
cndbcntrRef = signal->theData[0];
cmynodeid = signal->theData[1];
tstartphase = signal->theData[2];
tStartType = signal->theData[3];
switch (tstartphase) {
case ZSPH1:
jam();
ndbsttorryLab(signal);
return;
break;
case ZSPH2:
cnoLcpPages = 2 * (ZWRITEPAGESIZE + 1);
initialiseLcpPages(signal);
ndbsttorryLab(signal);
return;
break;
case ZSPH3:
if ((tStartType == NodeState::ST_NODE_RESTART) ||
(tStartType == NodeState::ST_INITIAL_NODE_RESTART)) {
jam();
//---------------------------------------------
// csystemRestart is used to check what is needed
// during log execution. When starting a node it
// is not a log execution and rather a normal
// execution. Thus we reset the variable here to
// avoid unnecessary system crashes.
//---------------------------------------------
csystemRestart = ZFALSE;
}//if
signal->theData[0] = ZLOAD_BAL_LCP_TIMER;
sendSignalWithDelay(cownBlockref, GSN_CONTINUEB, signal, 100, 1);
break;
case ZSPH6:
jam();
clblPagesPerTick = clblPagesPerTickAfterSr;
csystemRestart = ZFALSE;
signal->theData[0] = ZREPORT_MEMORY_USAGE;
sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, 2000, 1);
break;
default:
jam();
/*empty*/;
break;
}//switch
ndbsttorryLab(signal);
return;
}//Dbacc::execNDB_STTOR()
/* ******************--------------------------------------------------------------- */
/* STTOR START / RESTART */
/* ******************------------------------------+ */
/* SENDER: ANY, LEVEL B */
void Dbacc::execSTTOR(Signal* signal)
{
jamEntry();
// tstartphase = signal->theData[1];
tuserblockref = signal->theData[3];
csignalkey = signal->theData[6];
sttorrysignalLab(signal);
return;
}//Dbacc::execSTTOR()
/* --------------------------------------------------------------------------------- */
/* ZSPH1 */
/* --------------------------------------------------------------------------------- */
void Dbacc::ndbrestart1Lab(Signal* signal)
{
cmynodeid = globalData.ownId;
cownBlockref = numberToRef(DBACC, cmynodeid);
czero = 0;
cminusOne = czero - 1;
ctest = 0;
cundoLogActive = ZFALSE;
csystemRestart = ZTRUE;
clblPageOver = 0;
clblPageCounter = 0;
cactiveUndoFilePage = 0;
cprevUndoaddress = cminusOne;
cundoposition = 0;
clastUndoPageIdWritten = 0;
cactiveUndoFileVersion = RNIL;
cactiveOpenUndoFsPtr = RNIL;
for (Uint32 tmp = 0; tmp < ZMAX_UNDO_VERSION; tmp++) {
csrVersList[tmp] = RNIL;
}//for
return;
}//Dbacc::ndbrestart1Lab()
void Dbacc::initialiseRecordsLab(Signal* signal, Uint32 ref, Uint32 data)
{
switch (tdata0) {
case 0:
jam();
initialiseTableRec(signal);
break;
case 1:
jam();
initialiseFsConnectionRec(signal);
break;
case 2:
jam();
initialiseFsOpRec(signal);
break;
case 3:
jam();
initialiseLcpConnectionRec(signal);
break;
case 4:
jam();
initialiseDirRec(signal);
break;
case 5:
jam();
initialiseDirRangeRec(signal);
break;
case 6:
jam();
initialiseFragRec(signal);
break;
case 7:
jam();
initialiseOverflowRec(signal);
break;
case 8:
jam();
initialiseOperationRec(signal);
break;
case 9:
jam();
initialisePageRec(signal);
break;
case 10:
jam();
initialiseRootfragRec(signal);
break;
case 11:
jam();
initialiseScanRec(signal);
break;
case 12:
jam();
initialiseSrVerRec(signal);
{
ReadConfigConf * conf = (ReadConfigConf*)signal->getDataPtrSend();
conf->senderRef = reference();
conf->senderData = data;
sendSignal(ref, GSN_READ_CONFIG_CONF, signal,
ReadConfigConf::SignalLength, JBB);
}
return;
break;
default:
ndbrequire(false);
break;
}//switch
signal->theData[0] = ZINITIALISE_RECORDS;
signal->theData[1] = tdata0 + 1;
signal->theData[2] = 0;
signal->theData[3] = ref;
signal->theData[4] = data;
sendSignal(reference(), GSN_CONTINUEB, signal, 5, JBB);
return;
}//Dbacc::initialiseRecordsLab()
/* *********************************<< */
/* NDB_STTORRY */
/* *********************************<< */
void Dbacc::ndbsttorryLab(Signal* signal)
{
signal->theData[0] = cownBlockref;
sendSignal(cndbcntrRef, GSN_NDB_STTORRY, signal, 1, JBB);
return;
}//Dbacc::ndbsttorryLab()
/* *********************************<< */
/* SIZEALT_REP SIZE ALTERATION */
/* *********************************<< */
void Dbacc::execREAD_CONFIG_REQ(Signal* signal)
{
const ReadConfigReq * req = (ReadConfigReq*)signal->getDataPtr();
Uint32 ref = req->senderRef;
Uint32 senderData = req->senderData;
ndbrequire(req->noOfParameters == 0);
jamEntry();
const ndb_mgm_configuration_iterator * p =
theConfiguration.getOwnConfigIterator();
ndbrequire(p != 0);
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_ACC_DIR_RANGE, &cdirrangesize));
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_ACC_DIR_ARRAY, &cdirarraysize));
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_ACC_FRAGMENT, &cfragmentsize));
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_ACC_OP_RECS, &coprecsize));
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_ACC_OVERFLOW_RECS,
&coverflowrecsize));
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_ACC_PAGE8, &cpagesize));
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_ACC_ROOT_FRAG,
&crootfragmentsize));
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_ACC_TABLE, &ctablesize));
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_ACC_SCAN, &cscanRecSize));
initRecords();
ndbrestart1Lab(signal);
clblPagesPerTick = 50;
ndb_mgm_get_int_parameter(p, CFG_DB_LCP_DISC_PAGES_ACC_SR,
&clblPagesPerTick);
clblPagesPerTickAfterSr = 50;
ndb_mgm_get_int_parameter(p, CFG_DB_LCP_DISC_PAGES_ACC,
&clblPagesPerTickAfterSr);
tdata0 = 0;
initialiseRecordsLab(signal, ref, senderData);
return;
}//Dbacc::execSIZEALT_REP()
/* *********************************<< */
/* STTORRY */
/* *********************************<< */
void Dbacc::sttorrysignalLab(Signal* signal)
{
signal->theData[0] = csignalkey;
signal->theData[1] = 3;
/* BLOCK CATEGORY */
signal->theData[2] = 2;
/* SIGNAL VERSION NUMBER */
signal->theData[3] = ZSPH1;
signal->theData[4] = 255;
sendSignal(NDBCNTR_REF, GSN_STTORRY, signal, 5, JBB);
/* END OF START PHASES */
return;
}//Dbacc::sttorrysignalLab()
/* --------------------------------------------------------------------------------- */
/* INITIALISE_DIR_REC */
/* INITIALATES THE DIRECTORY RECORDS. */
/* --------------------------------------------------------------------------------- */
void Dbacc::initialiseDirRec(Signal* signal)
{
DirectoryarrayPtr idrDirptr;
ndbrequire(cdirarraysize > 0);
for (idrDirptr.i = 0; idrDirptr.i < cdirarraysize; idrDirptr.i++) {
refresh_watch_dog();
ptrAss(idrDirptr, directoryarray);
for (Uint32 i = 0; i <= 255; i++) {
idrDirptr.p->pagep[i] = RNIL;
}//for
}//for
cdirmemory = 0;
cfirstfreedir = RNIL;
}//Dbacc::initialiseDirRec()
/* --------------------------------------------------------------------------------- */
/* INITIALISE_DIR_RANGE_REC */
/* INITIALATES THE DIR_RANGE RECORDS. */
/* --------------------------------------------------------------------------------- */
void Dbacc::initialiseDirRangeRec(Signal* signal)
{
DirRangePtr idrDirRangePtr;
ndbrequire(cdirrangesize > 0);
for (idrDirRangePtr.i = 0; idrDirRangePtr.i < cdirrangesize; idrDirRangePtr.i++) {
refresh_watch_dog();
ptrAss(idrDirRangePtr, dirRange);
idrDirRangePtr.p->dirArray[0] = idrDirRangePtr.i + 1;
for (Uint32 i = 1; i < 256; i++) {
idrDirRangePtr.p->dirArray[i] = RNIL;
}//for
}//for
idrDirRangePtr.i = cdirrangesize - 1;
ptrAss(idrDirRangePtr, dirRange);
idrDirRangePtr.p->dirArray[0] = RNIL;
cfirstfreeDirrange = 0;
}//Dbacc::initialiseDirRangeRec()
/* --------------------------------------------------------------------------------- */
/* INITIALISE_FRAG_REC */
/* INITIALATES THE FRAGMENT RECORDS. */
/* --------------------------------------------------------------------------------- */
void Dbacc::initialiseFragRec(Signal* signal)
{
FragmentrecPtr regFragPtr;
ndbrequire(cfragmentsize > 0);
for (regFragPtr.i = 0; regFragPtr.i < cfragmentsize; regFragPtr.i++) {
jam();
refresh_watch_dog();
ptrAss(regFragPtr, fragmentrec);
initFragGeneral(regFragPtr);
regFragPtr.p->nextfreefrag = regFragPtr.i + 1;
}//for
regFragPtr.i = cfragmentsize - 1;
ptrAss(regFragPtr, fragmentrec);
regFragPtr.p->nextfreefrag = RNIL;
cfirstfreefrag = 0;
}//Dbacc::initialiseFragRec()
/* --------------------------------------------------------------------------------- */
/* INITIALISE_FS_CONNECTION_REC */
/* INITIALATES THE FS_CONNECTION RECORDS */
/* --------------------------------------------------------------------------------- */
void Dbacc::initialiseFsConnectionRec(Signal* signal)
{
ndbrequire(cfsConnectsize > 0);
for (fsConnectptr.i = 0; fsConnectptr.i < cfsConnectsize; fsConnectptr.i++) {
ptrAss(fsConnectptr, fsConnectrec);
fsConnectptr.p->fsNext = fsConnectptr.i + 1;
fsConnectptr.p->fsPrev = RNIL;
fsConnectptr.p->fragrecPtr = RNIL;
fsConnectptr.p->fsState = WAIT_NOTHING;
}//for
fsConnectptr.i = cfsConnectsize - 1;
ptrAss(fsConnectptr, fsConnectrec);
fsConnectptr.p->fsNext = RNIL; /* INITIALITES THE LAST CONNECTRECORD */
cfsFirstfreeconnect = 0; /* INITIATES THE FIRST FREE CONNECT RECORD */
}//Dbacc::initialiseFsConnectionRec()
/* --------------------------------------------------------------------------------- */
/* INITIALISE_FS_OP_REC */
/* INITIALATES THE FS_OP RECORDS */
/* --------------------------------------------------------------------------------- */
void Dbacc::initialiseFsOpRec(Signal* signal)
{
ndbrequire(cfsOpsize > 0);
for (fsOpptr.i = 0; fsOpptr.i < cfsOpsize; fsOpptr.i++) {
ptrAss(fsOpptr, fsOprec);
fsOpptr.p->fsOpnext = fsOpptr.i + 1;
fsOpptr.p->fsOpfragrecPtr = RNIL;
fsOpptr.p->fsConptr = RNIL;
fsOpptr.p->fsOpstate = WAIT_NOTHING;
}//for
fsOpptr.i = cfsOpsize - 1;
ptrAss(fsOpptr, fsOprec);
fsOpptr.p->fsOpnext = RNIL;
cfsFirstfreeop = 0;
}//Dbacc::initialiseFsOpRec()
/* --------------------------------------------------------------------------------- */
/* INITIALISE_LCP_CONNECTION_REC */
/* INITIALATES THE LCP_CONNECTION RECORDS */
/* --------------------------------------------------------------------------------- */
void Dbacc::initialiseLcpConnectionRec(Signal* signal)
{
ndbrequire(clcpConnectsize > 0);
for (lcpConnectptr.i = 0; lcpConnectptr.i < clcpConnectsize; lcpConnectptr.i++) {
ptrAss(lcpConnectptr, lcpConnectrec);
lcpConnectptr.p->nextLcpConn = lcpConnectptr.i + 1;
lcpConnectptr.p->lcpUserptr = RNIL;
lcpConnectptr.p->rootrecptr = RNIL;
lcpConnectptr.p->lcpstate = LCP_FREE;
}//for
lcpConnectptr.i = clcpConnectsize - 1;
ptrAss(lcpConnectptr, lcpConnectrec);
lcpConnectptr.p->nextLcpConn = RNIL;
cfirstfreelcpConnect = 0;
}//Dbacc::initialiseLcpConnectionRec()
/* --------------------------------------------------------------------------------- */
/* INITIALISE_OPERATION_REC */
/* INITIALATES THE OPERATION RECORDS. */
/* --------------------------------------------------------------------------------- */
void Dbacc::initialiseOperationRec(Signal* signal)
{
ndbrequire(coprecsize > 0);
for (operationRecPtr.i = 0; operationRecPtr.i < coprecsize; operationRecPtr.i++) {
refresh_watch_dog();
ptrAss(operationRecPtr, operationrec);
operationRecPtr.p->transactionstate = IDLE;
operationRecPtr.p->operation = ZUNDEFINED_OP;
operationRecPtr.p->opState = FREE_OP;
operationRecPtr.p->nextOp = operationRecPtr.i + 1;
}//for
operationRecPtr.i = coprecsize - 1;
ptrAss(operationRecPtr, operationrec);
operationRecPtr.p->nextOp = RNIL;
cfreeopRec = 0;
}//Dbacc::initialiseOperationRec()
/* --------------------------------------------------------------------------------- */
/* INITIALISE_OVERFLOW_REC */
/* INITIALATES THE OVERFLOW RECORDS */
/* --------------------------------------------------------------------------------- */
void Dbacc::initialiseOverflowRec(Signal* signal)
{
OverflowRecordPtr iorOverflowRecPtr;
ndbrequire(coverflowrecsize > 0);
for (iorOverflowRecPtr.i = 0; iorOverflowRecPtr.i < coverflowrecsize; iorOverflowRecPtr.i++) {
refresh_watch_dog();
ptrAss(iorOverflowRecPtr, overflowRecord);
iorOverflowRecPtr.p->nextfreeoverrec = iorOverflowRecPtr.i + 1;
}//for
iorOverflowRecPtr.i = coverflowrecsize - 1;
ptrAss(iorOverflowRecPtr, overflowRecord);
iorOverflowRecPtr.p->nextfreeoverrec = RNIL;
cfirstfreeoverrec = 0;
}//Dbacc::initialiseOverflowRec()
/* --------------------------------------------------------------------------------- */
/* INITIALISE_PAGE_REC */
/* INITIALATES THE PAGE RECORDS. */
/* --------------------------------------------------------------------------------- */
void Dbacc::initialisePageRec(Signal* signal)
{
ndbrequire(cpagesize > 0);
cfreepage = 0;
cfirstfreepage = RNIL;
cnoOfAllocatedPages = 0;
}//Dbacc::initialisePageRec()
/* --------------------------------------------------------------------------------- */
/* INITIALISE_LCP_PAGES */
/* INITIALATES THE LCP PAGE RECORDS. */
/* --------------------------------------------------------------------------------- */
void Dbacc::initialiseLcpPages(Signal* signal)
{
Uint32 tilpIndex;
ndbrequire(cnoLcpPages >= (2 * (ZWRITEPAGESIZE + 1)));
/* --------------------------------------------------------------------------------- */
/* AN ABSOLUTE MINIMUM IS THAT WE HAVE 16 LCP PAGES TO HANDLE TWO CONCURRENT */
/* LCP'S ON LOCAL FRAGMENTS. */
/* --------------------------------------------------------------------------------- */
ndbrequire(cpagesize >= (cnoLcpPages + 8));
/* --------------------------------------------------------------------------------- */
/* THE NUMBER OF PAGES MUST BE AT LEAST 8 PLUS THE NUMBER OF PAGES REQUIRED BY */
/* THE LOCAL CHECKPOINT PROCESS. THIS NUMBER IS 8 TIMES THE PARALLELISM OF */
/* LOCAL CHECKPOINTS. */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* WE SET UP A LINKED LIST OF PAGES FOR EXCLUSIVE USE BY LOCAL CHECKPOINTS. */
/* --------------------------------------------------------------------------------- */
cfirstfreeLcpPage = RNIL;
for (tilpIndex = 0; tilpIndex < cnoLcpPages; tilpIndex++) {
jam();
seizePage(signal);
rlpPageptr = spPageptr;
releaseLcpPage(signal);
}//for
}//Dbacc::initialiseLcpPages()
/* --------------------------------------------------------------------------------- */
/* INITIALISE_ROOTFRAG_REC */
/* INITIALATES THE ROOTFRAG RECORDS. */
/* --------------------------------------------------------------------------------- */
void Dbacc::initialiseRootfragRec(Signal* signal)
{
ndbrequire(crootfragmentsize > 0);
for (rootfragrecptr.i = 0; rootfragrecptr.i < crootfragmentsize; rootfragrecptr.i++) {
refresh_watch_dog();
ptrAss(rootfragrecptr, rootfragmentrec);
rootfragrecptr.p->nextroot = rootfragrecptr.i + 1;
rootfragrecptr.p->fragmentptr[0] = RNIL;
rootfragrecptr.p->fragmentptr[1] = RNIL;
}//for
rootfragrecptr.i = crootfragmentsize - 1;
ptrAss(rootfragrecptr, rootfragmentrec);
rootfragrecptr.p->nextroot = RNIL;
cfirstfreerootfrag = 0;
}//Dbacc::initialiseRootfragRec()
/* --------------------------------------------------------------------------------- */
/* INITIALISE_SCAN_REC */
/* INITIALATES THE QUE_SCAN RECORDS. */
/* --------------------------------------------------------------------------------- */
void Dbacc::initialiseScanRec(Signal* signal)
{
ndbrequire(cscanRecSize > 0);
for (scanPtr.i = 0; scanPtr.i < cscanRecSize; scanPtr.i++) {
ptrAss(scanPtr, scanRec);
scanPtr.p->scanNextfreerec = scanPtr.i + 1;
scanPtr.p->scanState = ScanRec::SCAN_DISCONNECT;
scanPtr.p->scanTimer = 0;
scanPtr.p->scanContinuebCounter = 0;
}//for
scanPtr.i = cscanRecSize - 1;
ptrAss(scanPtr, scanRec);
scanPtr.p->scanNextfreerec = RNIL;
cfirstFreeScanRec = 0;
}//Dbacc::initialiseScanRec()
/* --------------------------------------------------------------------------------- */
/* INITIALISE_SR_VER_REC */
/* --------------------------------------------------------------------------------- */
void Dbacc::initialiseSrVerRec(Signal* signal)
{
ndbrequire(csrVersionRecSize > 0);
for (srVersionPtr.i = 0; srVersionPtr.i < csrVersionRecSize; srVersionPtr.i++) {
ptrAss(srVersionPtr, srVersionRec);
srVersionPtr.p->nextFreeSr = srVersionPtr.i + 1;
}//for
srVersionPtr.i = csrVersionRecSize - 1;
ptrAss(srVersionPtr, srVersionRec);
srVersionPtr.p->nextFreeSr = RNIL;
cfirstFreeSrVersionRec = 0;
}//Dbacc::initialiseSrVerRec()
/* --------------------------------------------------------------------------------- */
/* INITIALISE_TABLE_REC */
/* INITIALATES THE TABLE RECORDS. */
/* --------------------------------------------------------------------------------- */
void Dbacc::initialiseTableRec(Signal* signal)
{
ndbrequire(ctablesize > 0);
for (tabptr.i = 0; tabptr.i < ctablesize; tabptr.i++) {
refresh_watch_dog();
ptrAss(tabptr, tabrec);
for (Uint32 i = 0; i < MAX_FRAG_PER_NODE; i++) {
tabptr.p->fragholder[i] = RNIL;
tabptr.p->fragptrholder[i] = RNIL;
}//for
}//for
}//Dbacc::initialiseTableRec()
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* */
/* END OF SYSTEM RESTART MODULE */
/* */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* */
/* ADD/DELETE FRAGMENT MODULE */
/* */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
void Dbacc::initRootfragrec(Signal* signal)
{
const AccFragReq * const req = (AccFragReq*)&signal->theData[0];
rootfragrecptr.p->mytabptr = req->tableId;
rootfragrecptr.p->roothashcheck = req->kValue + req->lhFragBits;
rootfragrecptr.p->noOfElements = 0;
rootfragrecptr.p->m_commit_count = 0;
for (Uint32 i = 0; i < MAX_PARALLEL_SCANS_PER_FRAG; i++) {
rootfragrecptr.p->scan[i] = RNIL;
}//for
}//Dbacc::initRootfragrec()
void Dbacc::execACCFRAGREQ(Signal* signal)
{
const AccFragReq * const req = (AccFragReq*)&signal->theData[0];
jamEntry();
if (ERROR_INSERTED(3001)) {
jam();
addFragRefuse(signal, 1);
CLEAR_ERROR_INSERT_VALUE;
return;
}
tabptr.i = req->tableId;
#ifndef VM_TRACE
// config mismatch - do not crash if release compiled
if (tabptr.i >= ctablesize) {
jam();
addFragRefuse(signal, 800);
return;
}
#endif
ptrCheckGuard(tabptr, ctablesize, tabrec);
ndbrequire((req->reqInfo & 0xF) == ZADDFRAG);
ndbrequire(!getrootfragmentrec(signal, rootfragrecptr, req->fragId));
if (cfirstfreerootfrag == RNIL) {
jam();
addFragRefuse(signal, ZFULL_ROOTFRAGRECORD_ERROR);
return;
}//if
seizeRootfragrec(signal);
if (!addfragtotab(signal, rootfragrecptr.i, req->fragId)) {
jam();
releaseRootFragRecord(signal, rootfragrecptr);
addFragRefuse(signal, ZFULL_ROOTFRAGRECORD_ERROR);
return;
}//if
initRootfragrec(signal);
for (Uint32 i = 0; i < 2; i++) {
jam();
if (cfirstfreefrag == RNIL) {
jam();
addFragRefuse(signal, ZFULL_FRAGRECORD_ERROR);
return;
}//if
seizeFragrec(signal);
initFragGeneral(fragrecptr);
initFragAdd(signal, i, rootfragrecptr.i, fragrecptr);
rootfragrecptr.p->fragmentptr[i] = fragrecptr.i;
rootfragrecptr.p->fragmentid[i] = fragrecptr.p->myfid;
if (cfirstfreeDirrange == RNIL) {
jam();
addFragRefuse(signal, ZDIR_RANGE_ERROR);
return;
} else {
jam();
seizeDirrange(signal);
}//if
fragrecptr.p->directory = newDirRangePtr.i;
seizeDirectory(signal);
if (tresult < ZLIMIT_OF_ERROR) {
jam();
newDirRangePtr.p->dirArray[0] = sdDirptr.i;
} else {
jam();
addFragRefuse(signal, tresult);
return;
}//if
seizePage(signal);
if (tresult > ZLIMIT_OF_ERROR) {
jam();
addFragRefuse(signal, tresult);
return;
}//if
sdDirptr.p->pagep[0] = spPageptr.i;
tipPageId = 0;
inpPageptr = spPageptr;
initPage(signal);
if (cfirstfreeDirrange == RNIL) {
jam();
addFragRefuse(signal, ZDIR_RANGE_ERROR);
return;
} else {
jam();
seizeDirrange(signal);
}//if
fragrecptr.p->overflowdir = newDirRangePtr.i;
seizeDirectory(signal);
if (tresult < ZLIMIT_OF_ERROR) {
jam();
newDirRangePtr.p->dirArray[0] = sdDirptr.i;
} else {
jam();
addFragRefuse(signal, tresult);
return;
}//if
}//for
Uint32 userPtr = req->userPtr;
BlockReference retRef = req->userRef;
rootfragrecptr.p->rootState = ACTIVEROOT;
AccFragConf * const conf = (AccFragConf*)&signal->theData[0];
conf->userPtr = userPtr;
conf->rootFragPtr = rootfragrecptr.i;
conf->fragId[0] = rootfragrecptr.p->fragmentid[0];
conf->fragId[1] = rootfragrecptr.p->fragmentid[1];
conf->fragPtr[0] = rootfragrecptr.p->fragmentptr[0];
conf->fragPtr[1] = rootfragrecptr.p->fragmentptr[1];
conf->rootHashCheck = rootfragrecptr.p->roothashcheck;
sendSignal(retRef, GSN_ACCFRAGCONF, signal, AccFragConf::SignalLength, JBB);
}//Dbacc::execACCFRAGREQ()
void Dbacc::addFragRefuse(Signal* signal, Uint32 errorCode)
{
const AccFragReq * const req = (AccFragReq*)&signal->theData[0];
AccFragRef * const ref = (AccFragRef*)&signal->theData[0];
Uint32 userPtr = req->userPtr;
BlockReference retRef = req->userRef;
ref->userPtr = userPtr;
ref->errorCode = errorCode;
sendSignal(retRef, GSN_ACCFRAGREF, signal, AccFragRef::SignalLength, JBB);
return;
}//Dbacc::addFragRefuseEarly()
void
Dbacc::execDROP_TAB_REQ(Signal* signal){
jamEntry();
DropTabReq* req = (DropTabReq*)signal->getDataPtr();
TabrecPtr tabPtr;
tabPtr.i = req->tableId;
ptrCheckGuard(tabPtr, ctablesize, tabrec);
tabPtr.p->tabUserRef = req->senderRef;
tabPtr.p->tabUserPtr = req->senderData;
signal->theData[0] = ZREL_ROOT_FRAG;
signal->theData[1] = tabPtr.i;
sendSignal(cownBlockref, GSN_CONTINUEB, signal, 2, JBB);
}
void Dbacc::releaseRootFragResources(Signal* signal, Uint32 tableId)
{
RootfragmentrecPtr rootPtr;
TabrecPtr tabPtr;
tabPtr.i = tableId;
ptrCheckGuard(tabPtr, ctablesize, tabrec);
for (Uint32 i = 0; i < MAX_FRAG_PER_NODE; i++) {
jam();
if (tabPtr.p->fragholder[i] != RNIL) {
jam();
Uint32 fragIndex;
rootPtr.i = tabPtr.p->fragptrholder[i];
ptrCheckGuard(rootPtr, crootfragmentsize, rootfragmentrec);
if (rootPtr.p->fragmentptr[0] != RNIL) {
jam();
fragIndex = rootPtr.p->fragmentptr[0];
rootPtr.p->fragmentptr[0] = RNIL;
} else if (rootPtr.p->fragmentptr[1] != RNIL) {
jam();
fragIndex = rootPtr.p->fragmentptr[1];
rootPtr.p->fragmentptr[1] = RNIL;
} else {
jam();
releaseRootFragRecord(signal, rootPtr);
tabPtr.p->fragholder[i] = RNIL;
tabPtr.p->fragptrholder[i] = RNIL;
continue;
}//if
releaseFragResources(signal, fragIndex);
return;
}//if
}//for
/**
* Finished...
*/
sendFSREMOVEREQ(signal, tableId);
}//Dbacc::releaseRootFragResources()
void Dbacc::releaseRootFragRecord(Signal* signal, RootfragmentrecPtr rootPtr)
{
rootPtr.p->nextroot = cfirstfreerootfrag;
cfirstfreerootfrag = rootPtr.i;
}//Dbacc::releaseRootFragRecord()
void Dbacc::releaseFragResources(Signal* signal, Uint32 fragIndex)
{
FragmentrecPtr regFragPtr;
regFragPtr.i = fragIndex;
ptrCheckGuard(regFragPtr, cfragmentsize, fragmentrec);
verifyFragCorrect(regFragPtr);
if (regFragPtr.p->directory != RNIL) {
jam();
releaseDirResources(signal, regFragPtr.i, regFragPtr.p->directory, 0);
regFragPtr.p->directory = RNIL;
} else if (regFragPtr.p->overflowdir != RNIL) {
jam();
releaseDirResources(signal, regFragPtr.i, regFragPtr.p->overflowdir, 0);
regFragPtr.p->overflowdir = RNIL;
} else if (regFragPtr.p->firstOverflowRec != RNIL) {
jam();
releaseOverflowResources(signal, regFragPtr);
} else if (regFragPtr.p->firstFreeDirindexRec != RNIL) {
jam();
releaseDirIndexResources(signal, regFragPtr);
} else {
RootfragmentrecPtr rootPtr;
jam();
rootPtr.i = regFragPtr.p->myroot;
ptrCheckGuard(rootPtr, crootfragmentsize, rootfragmentrec);
releaseFragRecord(signal, regFragPtr);
signal->theData[0] = ZREL_ROOT_FRAG;
signal->theData[1] = rootPtr.p->mytabptr;
sendSignal(cownBlockref, GSN_CONTINUEB, signal, 2, JBB);
}//if
}//Dbacc::releaseFragResources()
void Dbacc::verifyFragCorrect(FragmentrecPtr regFragPtr)
{
for (Uint32 i = 0; i < ZWRITEPAGESIZE; i++) {
jam();
ndbrequire(regFragPtr.p->datapages[i] == RNIL);
}//for
ndbrequire(regFragPtr.p->lockOwnersList == RNIL);
ndbrequire(regFragPtr.p->firstWaitInQueOp == RNIL);
ndbrequire(regFragPtr.p->lastWaitInQueOp == RNIL);
ndbrequire(regFragPtr.p->sentWaitInQueOp == RNIL);
//ndbrequire(regFragPtr.p->fsConnPtr == RNIL);
ndbrequire(regFragPtr.p->zeroPagePtr == RNIL);
ndbrequire(regFragPtr.p->nrWaitWriteUndoExit == 0);
ndbrequire(regFragPtr.p->sentWaitInQueOp == RNIL);
}//Dbacc::verifyFragCorrect()
void Dbacc::releaseDirResources(Signal* signal,
Uint32 fragIndex,
Uint32 dirIndex,
Uint32 startIndex)
{
DirRangePtr regDirRangePtr;
regDirRangePtr.i = dirIndex;
ptrCheckGuard(regDirRangePtr, cdirrangesize, dirRange);
for (Uint32 i = startIndex; i < 256; i++) {
jam();
if (regDirRangePtr.p->dirArray[i] != RNIL) {
jam();
Uint32 directoryIndex = regDirRangePtr.p->dirArray[i];
regDirRangePtr.p->dirArray[i] = RNIL;
releaseDirectoryResources(signal, fragIndex, dirIndex, (i + 1), directoryIndex);
return;
}//if
}//for
rdDirRangePtr = regDirRangePtr;
releaseDirrange(signal);
signal->theData[0] = ZREL_FRAG;
signal->theData[1] = fragIndex;
sendSignal(cownBlockref, GSN_CONTINUEB, signal, 2, JBB);
}//Dbacc::releaseDirResources()
void Dbacc::releaseDirectoryResources(Signal* signal,
Uint32 fragIndex,
Uint32 dirIndex,
Uint32 startIndex,
Uint32 directoryIndex)
{
DirectoryarrayPtr regDirPtr;
regDirPtr.i = directoryIndex;
ptrCheckGuard(regDirPtr, cdirarraysize, directoryarray);
for (Uint32 i = 0; i < 256; i++) {
jam();
if (regDirPtr.p->pagep[i] != RNIL) {
jam();
rpPageptr.i = regDirPtr.p->pagep[i];
ptrCheckGuard(rpPageptr, cpagesize, page8);
releasePage(signal);
regDirPtr.p->pagep[i] = RNIL;
}//if
}//for
rdDirptr = regDirPtr;
releaseDirectory(signal);
signal->theData[0] = ZREL_DIR;
signal->theData[1] = fragIndex;
signal->theData[2] = dirIndex;
signal->theData[3] = startIndex;
sendSignal(cownBlockref, GSN_CONTINUEB, signal, 4, JBB);
}//Dbacc::releaseDirectoryResources()
void Dbacc::releaseOverflowResources(Signal* signal, FragmentrecPtr regFragPtr)
{
Uint32 loopCount = 0;
OverflowRecordPtr regOverflowRecPtr;
while ((regFragPtr.p->firstOverflowRec != RNIL) &&
(loopCount < 1)) {
jam();
regOverflowRecPtr.i = regFragPtr.p->firstOverflowRec;
ptrCheckGuard(regOverflowRecPtr, coverflowrecsize, overflowRecord);
regFragPtr.p->firstOverflowRec = regOverflowRecPtr.p->nextOverRec;
rorOverflowRecPtr = regOverflowRecPtr;
releaseOverflowRec(signal);
loopCount++;
}//while
signal->theData[0] = ZREL_FRAG;
signal->theData[1] = regFragPtr.i;
sendSignal(cownBlockref, GSN_CONTINUEB, signal, 2, JBB);
}//Dbacc::releaseOverflowResources()
void Dbacc::releaseDirIndexResources(Signal* signal, FragmentrecPtr regFragPtr)
{
Uint32 loopCount = 0;
OverflowRecordPtr regOverflowRecPtr;
while ((regFragPtr.p->firstFreeDirindexRec != RNIL) &&
(loopCount < 1)) {
jam();
regOverflowRecPtr.i = regFragPtr.p->firstFreeDirindexRec;
ptrCheckGuard(regOverflowRecPtr, coverflowrecsize, overflowRecord);
regFragPtr.p->firstFreeDirindexRec = regOverflowRecPtr.p->nextOverList;
rorOverflowRecPtr = regOverflowRecPtr;
releaseOverflowRec(signal);
loopCount++;
}//while
signal->theData[0] = ZREL_FRAG;
signal->theData[1] = regFragPtr.i;
sendSignal(cownBlockref, GSN_CONTINUEB, signal, 2, JBB);
}//Dbacc::releaseDirIndexResources()
void Dbacc::releaseFragRecord(Signal* signal, FragmentrecPtr regFragPtr)
{
regFragPtr.p->nextfreefrag = cfirstfreefrag;
cfirstfreefrag = regFragPtr.i;
initFragGeneral(regFragPtr);
}//Dbacc::releaseFragRecord()
void Dbacc::sendFSREMOVEREQ(Signal* signal, Uint32 tableId)
{
FsRemoveReq * const fsReq = (FsRemoveReq *)signal->getDataPtrSend();
fsReq->userReference = cownBlockref;
fsReq->userPointer = tableId;
fsReq->fileNumber[0] = tableId;
fsReq->fileNumber[1] = (Uint32)-1; // Remove all fragments
fsReq->fileNumber[2] = (Uint32)-1; // Remove all data files within fragment
fsReq->fileNumber[3] = 255 | // No P-value used here
(3 << 8) | // Data-files in D3
(0 << 16) | // Data-files
(1 << 24); // Version 1 of fileNumber
fsReq->directory = 1;
fsReq->ownDirectory = 1;
sendSignal(NDBFS_REF, GSN_FSREMOVEREQ, signal, FsRemoveReq::SignalLength, JBA);
}//Dbacc::sendFSREMOVEREQ()
void Dbacc::execFSREMOVECONF(Signal* signal)
{
FsConf * const fsConf = (FsConf *)signal->getDataPtrSend();
TabrecPtr tabPtr;
tabPtr.i = fsConf->userPointer;
ptrCheckGuard(tabPtr, ctablesize, tabrec);
DropTabConf * const dropConf = (DropTabConf *)signal->getDataPtrSend();
dropConf->senderRef = reference();
dropConf->senderData = tabPtr.p->tabUserPtr;
dropConf->tableId = tabPtr.i;
sendSignal(tabPtr.p->tabUserRef, GSN_DROP_TAB_CONF,
signal, DropTabConf::SignalLength, JBB);
tabPtr.p->tabUserPtr = RNIL;
tabPtr.p->tabUserRef = 0;
}//Dbacc::execFSREMOVECONF()
/* -------------------------------------------------------------------------- */
/* ADDFRAGTOTAB */
/* DESCRIPTION: PUTS A FRAGMENT ID AND A POINTER TO ITS RECORD INTO */
/* TABLE ARRRAY OF THE TABLE RECORD. */
/* -------------------------------------------------------------------------- */
bool Dbacc::addfragtotab(Signal* signal, Uint32 rootIndex, Uint32 fid)
{
for (Uint32 i = 0; i < MAX_FRAG_PER_NODE; i++) {
jam();
if (tabptr.p->fragholder[i] == RNIL) {
jam();
tabptr.p->fragholder[i] = fid;
tabptr.p->fragptrholder[i] = rootIndex;
return true;
}//if
}//for
return false;
}//Dbacc::addfragtotab()
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* */
/* END OF ADD/DELETE FRAGMENT MODULE */
/* */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* */
/* CONNECTION MODULE */
/* */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* ******************--------------------------------------------------------------- */
/* ACCSEIZEREQ SEIZE REQ */
/* SENDER: LQH, LEVEL B */
/* ENTER ACCSEIZEREQ WITH */
/* TUSERPTR , CONECTION PTR OF LQH */
/* TUSERBLOCKREF BLOCK REFERENCE OF LQH */
/* ******************--------------------------------------------------------------- */
/* ******************--------------------------------------------------------------- */
/* ACCSEIZEREQ SEIZE REQ */
/* ******************------------------------------+ */
/* SENDER: LQH, LEVEL B */
void Dbacc::execACCSEIZEREQ(Signal* signal)
{
jamEntry();
tuserptr = signal->theData[0];
/* CONECTION PTR OF LQH */
tuserblockref = signal->theData[1];
/* BLOCK REFERENCE OF LQH */
tresult = 0;
if (cfreeopRec == RNIL) {
jam();
refaccConnectLab(signal);
return;
}//if
seizeOpRec(signal);
ptrGuard(operationRecPtr);
operationRecPtr.p->userptr = tuserptr;
operationRecPtr.p->userblockref = tuserblockref;
operationRecPtr.p->operation = ZUNDEFINED_OP;
operationRecPtr.p->transactionstate = IDLE;
/* ******************************< */
/* ACCSEIZECONF */
/* ******************************< */
signal->theData[0] = tuserptr;
signal->theData[1] = operationRecPtr.i;
sendSignal(tuserblockref, GSN_ACCSEIZECONF, signal, 2, JBB);
return;
}//Dbacc::execACCSEIZEREQ()
void Dbacc::refaccConnectLab(Signal* signal)
{
tresult = ZCONNECT_SIZE_ERROR;
/* ******************************< */
/* ACCSEIZEREF */
/* ******************************< */
signal->theData[0] = tuserptr;
signal->theData[1] = tresult;
sendSignal(tuserblockref, GSN_ACCSEIZEREF, signal, 2, JBB);
return;
}//Dbacc::refaccConnectLab()
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* */
/* END OF CONNECTION MODULE */
/* */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* */
/* EXECUTE OPERATION MODULE */
/* */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* INIT_OP_REC */
/* INFORMATION WHICH IS RECIEVED BY ACCKEYREQ WILL BE SAVED */
/* IN THE OPERATION RECORD. */
/* --------------------------------------------------------------------------------- */
void Dbacc::initOpRec(Signal* signal)
{
register Uint32 Treqinfo;
Treqinfo = signal->theData[2];
operationRecPtr.p->hashValue = signal->theData[3];
operationRecPtr.p->tupkeylen = signal->theData[4];
operationRecPtr.p->transId1 = signal->theData[5];
operationRecPtr.p->transId2 = signal->theData[6];
operationRecPtr.p->transactionstate = ACTIVE;
operationRecPtr.p->commitDeleteCheckFlag = ZFALSE;
operationRecPtr.p->operation = Treqinfo & 0x7;
/* --------------------------------------------------------------------------------- */
// opSimple is not used in this version. Is needed for deadlock handling later on.
/* --------------------------------------------------------------------------------- */
// operationRecPtr.p->opSimple = (Treqinfo >> 3) & 0x1;
operationRecPtr.p->lockMode = (Treqinfo >> 4) & 0x3;
Uint32 readFlag = (((Treqinfo >> 4) & 0x3) == 0); // Only 1 if Read
Uint32 dirtyFlag = (((Treqinfo >> 6) & 0x1) == 1); // Only 1 if Dirty
Uint32 dirtyReadFlag = readFlag & dirtyFlag;
operationRecPtr.p->dirtyRead = dirtyReadFlag;
operationRecPtr.p->nodeType = (Treqinfo >> 7) & 0x3;
operationRecPtr.p->fid = fragrecptr.p->myfid;
operationRecPtr.p->fragptr = fragrecptr.i;
operationRecPtr.p->nextParallelQue = RNIL;
operationRecPtr.p->prevParallelQue = RNIL;
operationRecPtr.p->prevQueOp = RNIL;
operationRecPtr.p->nextQueOp = RNIL;
operationRecPtr.p->nextSerialQue = RNIL;
operationRecPtr.p->prevSerialQue = RNIL;
operationRecPtr.p->elementPage = RNIL;
operationRecPtr.p->keyinfoPage = RNIL;
operationRecPtr.p->lockOwner = ZFALSE;
operationRecPtr.p->insertIsDone = ZFALSE;
operationRecPtr.p->elementIsDisappeared = ZFALSE;
operationRecPtr.p->insertDeleteLen = fragrecptr.p->elementLength;
operationRecPtr.p->longPagePtr = RNIL;
operationRecPtr.p->longKeyPageIndex = RNIL;
operationRecPtr.p->scanRecPtr = RNIL;
// bit to mark lock operation
operationRecPtr.p->isAccLockReq = (Treqinfo >> 31) & 0x1;
}//Dbacc::initOpRec()
/* --------------------------------------------------------------------------------- */
/* SEND_ACCKEYCONF */
/* --------------------------------------------------------------------------------- */
void Dbacc::sendAcckeyconf(Signal* signal)
{
signal->theData[0] = operationRecPtr.p->userptr;
signal->theData[1] = operationRecPtr.p->operation;
signal->theData[2] = operationRecPtr.p->fid;
signal->theData[3] = operationRecPtr.p->localdata[0];
signal->theData[4] = operationRecPtr.p->localdata[1];
signal->theData[5] = fragrecptr.p->localkeylen;
}//Dbacc::sendAcckeyconf()
void Dbacc::ACCKEY_error(Uint32 fromWhere)
{
switch(fromWhere) {
case 0:
ndbrequire(false);
case 1:
ndbrequire(false);
case 2:
ndbrequire(false);
case 3:
ndbrequire(false);
case 4:
ndbrequire(false);
case 5:
ndbrequire(false);
case 6:
ndbrequire(false);
case 7:
ndbrequire(false);
case 8:
ndbrequire(false);
case 9:
ndbrequire(false);
default:
ndbrequire(false);
}//switch
}//Dbacc::ACCKEY_error()
/* ******************--------------------------------------------------------------- */
/* ACCKEYREQ REQUEST FOR INSERT, DELETE, */
/* RERAD AND UPDATE, A TUPLE. */
/* SENDER: LQH, LEVEL B */
/* SIGNAL DATA: OPERATION_REC_PTR, CONNECTION PTR */
/* TABPTR, TABLE ID = TABLE RECORD POINTER */
/* TREQINFO, */
/* THASHVALUE, HASH VALUE OF THE TUP */
/* TKEYLEN, LENGTH OF THE PRIMARY KEYS */
/* TKEY1, PRIMARY KEY 1 */
/* TKEY2, PRIMARY KEY 2 */
/* TKEY3, PRIMARY KEY 3 */
/* TKEY4, PRIMARY KEY 4 */
/* ******************--------------------------------------------------------------- */
void Dbacc::execACCKEYREQ(Signal* signal)
{
jamEntry();
operationRecPtr.i = signal->theData[0]; /* CONNECTION PTR */
fragrecptr.i = signal->theData[1]; /* FRAGMENT RECORD POINTER */
if (!((operationRecPtr.i < coprecsize) ||
(fragrecptr.i < cfragmentsize))) {
ACCKEY_error(0);
return;
}//if
ptrAss(operationRecPtr, operationrec);
ptrAss(fragrecptr, fragmentrec);
ndbrequire(operationRecPtr.p->transactionstate == IDLE);
initOpRec(signal);
/*---------------------------------------------------------------*/
/* */
/* WE WILL USE THE HASH VALUE TO LOOK UP THE PROPER MEMORY */
/* PAGE AND MEMORY PAGE INDEX TO START THE SEARCH WITHIN. */
/* WE REMEMBER THESE ADDRESS IF WE LATER NEED TO INSERT */
/* THE ITEM AFTER NOT FINDING THE ITEM. */
/*---------------------------------------------------------------*/
getElement(signal);
if (tgeResult == ZTRUE) {
switch (operationRecPtr.p->operation) {
case ZREAD:
case ZUPDATE:
case ZDELETE:
case ZWRITE:
case ZSCAN_OP:
if (!tgeLocked){
if(operationRecPtr.p->operation == ZWRITE)
{
jam();
operationRecPtr.p->operation = ZUPDATE;
}
sendAcckeyconf(signal);
if (operationRecPtr.p->dirtyRead == ZFALSE) {
/*---------------------------------------------------------------*/
// It is not a dirty read. We proceed by locking and continue with
// the operation.
/*---------------------------------------------------------------*/
Uint32 eh = gePageptr.p->word32[tgeElementptr];
operationRecPtr.p->scanBits = ElementHeader::getScanBits(eh);
operationRecPtr.p->hashvaluePart = ElementHeader::getHashValuePart(eh);
operationRecPtr.p->elementPage = gePageptr.i;
operationRecPtr.p->elementContainer = tgeContainerptr;
operationRecPtr.p->elementPointer = tgeElementptr;
operationRecPtr.p->elementIsforward = tgeForward;
eh = ElementHeader::setLocked(operationRecPtr.i);
dbgWord32(gePageptr, tgeElementptr, eh);
gePageptr.p->word32[tgeElementptr] = eh;
insertLockOwnersList(signal , operationRecPtr);
return;
} else {
jam();
/*---------------------------------------------------------------*/
// It is a dirty read. We do not lock anything. Set state to
// IDLE since no COMMIT call will come.
/*---------------------------------------------------------------*/
operationRecPtr.p->transactionstate = IDLE;
operationRecPtr.p->operation = ZUNDEFINED_OP;
return;
}//if
} else {
jam();
accIsLockedLab(signal);
return;
}//if
break;
case ZINSERT:
jam();
insertExistElemLab(signal);
return;
break;
default:
ndbrequire(false);
break;
}//switch
} else if (tgeResult == ZFALSE) {
switch (operationRecPtr.p->operation) {
case ZINSERT:
case ZWRITE:
jam();
// If a write operation makes an insert we switch operation to ZINSERT so
// that the commit-method knows an insert has been made and updates noOfElements.
operationRecPtr.p->operation = ZINSERT;
operationRecPtr.p->insertIsDone = ZTRUE;
insertelementLab(signal);
return;
break;
case ZREAD:
case ZUPDATE:
case ZDELETE:
case ZSCAN_OP:
jam();
acckeyref1Lab(signal, ZREAD_ERROR);
return;
break;
default:
ndbrequire(false);
break;
}//switch
} else {
jam();
acckeyref1Lab(signal, tgeResult);
return;
}//if
return;
}//Dbacc::execACCKEYREQ()
void Dbacc::accIsLockedLab(Signal* signal)
{
ndbrequire(csystemRestart == ZFALSE);
queOperPtr.i = ElementHeader::getOpPtrI(gePageptr.p->word32[tgeElementptr]);
ptrCheckGuard(queOperPtr, coprecsize, operationrec);
if (operationRecPtr.p->dirtyRead == ZFALSE) {
Uint32 return_result;
if (operationRecPtr.p->lockMode == ZREADLOCK) {
jam();
priPageptr = gePageptr;
tpriElementptr = tgeElementptr;
return_result = placeReadInLockQueue(signal);
} else {
jam();
pwiPageptr = gePageptr;
tpwiElementptr = tgeElementptr;
return_result = placeWriteInLockQueue(signal);
}//if
if (return_result == ZPARALLEL_QUEUE) {
jam();
sendAcckeyconf(signal);
return;
} else if (return_result == ZSERIAL_QUEUE) {
jam();
signal->theData[0] = RNIL;
return;
} else if (return_result == ZWRITE_ERROR) {
jam();
acckeyref1Lab(signal, return_result);
return;
}//if
ndbrequire(false);
} else {
if (queOperPtr.p->elementIsDisappeared == ZFALSE) {
jam();
/*---------------------------------------------------------------*/
// It is a dirty read. We do not lock anything. Set state to
// IDLE since no COMMIT call will arrive.
/*---------------------------------------------------------------*/
sendAcckeyconf(signal);
operationRecPtr.p->transactionstate = IDLE;
operationRecPtr.p->operation = ZUNDEFINED_OP;
return;
} else {
jam();
/*---------------------------------------------------------------*/
// The tuple does not exist in the committed world currently.
// Report read error.
/*---------------------------------------------------------------*/
acckeyref1Lab(signal, ZREAD_ERROR);
return;
}//if
}//if
}//Dbacc::accIsLockedLab()
/* --------------------------------------------------------------------------------- */
/* I N S E R T E X I S T E L E M E N T */
/* --------------------------------------------------------------------------------- */
void Dbacc::insertExistElemLab(Signal* signal)
{
if (!tgeLocked){
jam();
acckeyref1Lab(signal, ZWRITE_ERROR);/* THE ELEMENT ALREADY EXIST */
return;
}//if
accIsLockedLab(signal);
}//Dbacc::insertExistElemLab()
/* --------------------------------------------------------------------------------- */
/* INSERTELEMENT */
/* --------------------------------------------------------------------------------- */
void Dbacc::insertelementLab(Signal* signal)
{
Uint32 tinsKeyLen;
if (fragrecptr.p->createLcp == ZTRUE) {
if (remainingUndoPages() < ZMIN_UNDO_PAGES_AT_OPERATION) {
jam();
acckeyref1Lab(signal, ZTEMPORARY_ACC_UNDO_FAILURE);
return;
}//if
}//if
if (fragrecptr.p->firstOverflowRec == RNIL) {
jam();
allocOverflowPage(signal);
if (tresult > ZLIMIT_OF_ERROR) {
jam();
acckeyref1Lab(signal, tresult);
return;
}//if
}//if
if (fragrecptr.p->keyLength != operationRecPtr.p->tupkeylen) {
ndbrequire(fragrecptr.p->keyLength == 0);
}//if
if (fragrecptr.p->keyLength != 0) {
ndbrequire(operationRecPtr.p->tupkeylen <= 8);
for (Uint32 i = 0; i < operationRecPtr.p->tupkeylen; i++) {
jam();
ckeys[i] = signal->theData[i + 7];
}//for
tinsKeyLen = operationRecPtr.p->tupkeylen;
} else {
jam();
seizePage(signal);
if (tresult > ZLIMIT_OF_ERROR) {
jam();
acckeyref1Lab(signal, tresult);
return;
}//if
operationRecPtr.p->keyinfoPage = spPageptr.i;
for (Uint32 i = 0; i < signal->theData[4]; i++) {
spPageptr.p->word32[i] = signal->theData[i + 7];
}//for
getLongKeyPage(signal);
if (tresult > ZLIMIT_OF_ERROR) {
jam();
acckeyref1Lab(signal, tresult);
return;
}//if
slkPageptr = glkPageptr;
slkCopyPageptr.i = operationRecPtr.p->keyinfoPage;
ptrCheckGuard(slkCopyPageptr, cpagesize, page8);
tslkKeyLen = operationRecPtr.p->tupkeylen;
storeLongKeys(signal);
ckeys[0] = (slkPageptr.p->word32[ZPOS_PAGE_ID] << 10) + tslkPageIndex;
tinsKeyLen = ZACTIVE_LONG_KEY_LEN;
rpPageptr.i = operationRecPtr.p->keyinfoPage;
ptrCheckGuard(rpPageptr, cpagesize, page8);
releasePage(signal);
operationRecPtr.p->keyinfoPage = RNIL;
}//if
signal->theData[0] = operationRecPtr.p->userptr;
Uint32 blockNo = refToBlock(operationRecPtr.p->userblockref);
EXECUTE_DIRECT(blockNo, GSN_LQH_ALLOCREQ, signal, 1);
jamEntry();
if (signal->theData[0] != 0) {
jam();
Uint32 result_code = signal->theData[0];
acckeyref1Lab(signal, result_code);
return;
}//if
Uint32 localKey = (signal->theData[1] << MAX_TUPLES_BITS) + signal->theData[2];
insertLockOwnersList(signal, operationRecPtr);
const Uint32 tmp = fragrecptr.p->k + fragrecptr.p->lhfragbits;
operationRecPtr.p->hashvaluePart =
(operationRecPtr.p->hashValue >> tmp) & 0xFFFF;
operationRecPtr.p->scanBits = 0; /* NOT ANY ACTIVE SCAN */
tidrElemhead = ElementHeader::setLocked(operationRecPtr.i);
idrPageptr = gdiPageptr;
tidrPageindex = tgdiPageindex;
tidrForward = ZTRUE;
tidrKeyLen = tinsKeyLen;
idrOperationRecPtr = operationRecPtr;
clocalkey[0] = localKey;
operationRecPtr.p->localdata[0] = localKey;
/* --------------------------------------------------------------------------------- */
/* WE SET THE LOCAL KEY TO MINUS ONE TO INDICATE IT IS NOT YET VALID. */
/* --------------------------------------------------------------------------------- */
insertElement(signal);
sendAcckeyconf(signal);
return;
}//Dbacc::insertelementLab()
/* --------------------------------------------------------------------------------- */
/* PLACE_READ_IN_LOCK_QUEUE */
/* INPUT: OPERATION_REC_PTR OUR OPERATION POINTER */
/* QUE_OPER_PTR LOCK QUEUE OWNER OPERATION POINTER */
/* PRI_PAGEPTR PAGE POINTER OF ELEMENT */
/* TPRI_ELEMENTPTR ELEMENT POINTER OF ELEMENT */
/* OUTPUT TRESULT = */
/* ZPARALLEL_QUEUE OPERATION PLACED IN PARALLEL QUEUE */
/* OPERATION CAN PROCEED NOW. */
/* ZSERIAL_QUEUE OPERATION PLACED IN SERIAL QUEUE */
/* ERROR CODE OPERATION NEEDS ABORTING */
/* THE ELEMENT WAS LOCKED AND WE WANT TO READ THE TUPLE. WE WILL CHECK THE LOCK */
/* QUEUES TO PERFORM THE PROPER ACTION. */
/* */
/* IN SOME PLACES IN THE CODE BELOW THAT HANDLES WHAT TO DO WHEN THE TUPLE IS LOCKED */
/* WE DO ASSUME THAT NEXT_PARALLEL_QUEUE AND NEXT_SERIAL_QUEUE ON OPERATION_REC_PTR */
/* HAVE BEEN INITIALISED TO RNIL. THUS WE DO NOT PERFORM THIS ONCE MORE EVEN IF IT */
/* COULD BE NICE FOR READABILITY. */
/* --------------------------------------------------------------------------------- */
Uint32 Dbacc::placeReadInLockQueue(Signal* signal)
{
if (getNoParallelTransaction(queOperPtr.p) == 1) {
if ((queOperPtr.p->transId1 == operationRecPtr.p->transId1) &&
(queOperPtr.p->transId2 == operationRecPtr.p->transId2)) {
/* --------------------------------------------------------------------------------- */
/* WE ARE PERFORMING A READ OPERATION AND THIS TRANSACTION ALREADY OWNS THE LOCK */
/* ALONE. PUT THE OPERATION LAST IN THE PARALLEL QUEUE. */
/* --------------------------------------------------------------------------------- */
jam();
mlpqOperPtr = queOperPtr;
moveLastParallelQueue(signal);
operationRecPtr.p->localdata[0] = queOperPtr.p->localdata[0];
operationRecPtr.p->localdata[1] = queOperPtr.p->localdata[1];
operationRecPtr.p->prevParallelQue = mlpqOperPtr.i;
mlpqOperPtr.p->nextParallelQue = operationRecPtr.i;
switch (queOperPtr.p->lockMode) {
case ZREADLOCK:
jam();
/*empty*/;
break;
default:
jam();
/* --------------------------------------------------------------------------------- */
/* IF THE TRANSACTION PREVIOUSLY SET A WRITE LOCK WE MUST ENSURE THAT ALL */
/* OPERATIONS IN THE PARALLEL QUEUE HAVE WRITE LOCK MODE TO AVOID STRANGE BUGS.*/
/* --------------------------------------------------------------------------------- */
operationRecPtr.p->lockMode = queOperPtr.p->lockMode;
break;
}//switch
return ZPARALLEL_QUEUE;
}//if
}//if
if (queOperPtr.p->nextSerialQue == RNIL) {
/* --------------------------------------------------------------------------------- */
/* WE ARE PERFORMING A READ OPERATION AND THERE IS NO SERIAL QUEUE. IF THERE IS NO */
/* WRITE OPERATION THAT OWNS THE LOCK OR ANY WRITE OPERATION IN THE PARALLEL QUEUE */
/* IT IS ENOUGH TO CHECK THE LOCK MODE OF THE LEADER IN THE PARALLEL QUEUE. IF IT IS */
/* A READ LOCK THEN WE PLACE OURSELVES IN THE PARALLEL QUEUE OTHERWISE WE GO ON TO */
/* PLACE OURSELVES IN THE SERIAL QUEUE. */
/* --------------------------------------------------------------------------------- */
switch (queOperPtr.p->lockMode) {
case ZREADLOCK:
jam();
mlpqOperPtr = queOperPtr;
moveLastParallelQueue(signal);
operationRecPtr.p->prevParallelQue = mlpqOperPtr.i;
mlpqOperPtr.p->nextParallelQue = operationRecPtr.i;
operationRecPtr.p->localdata[0] = queOperPtr.p->localdata[0];
operationRecPtr.p->localdata[1] = queOperPtr.p->localdata[1];
return ZPARALLEL_QUEUE;
default:
jam();
queOperPtr.p->nextSerialQue = operationRecPtr.i;
operationRecPtr.p->prevSerialQue = queOperPtr.i;
putOpInFragWaitQue(signal);
break;
}//switch
} else {
jam();
placeSerialQueueRead(signal);
}//if
return ZSERIAL_QUEUE;
}//Dbacc::placeReadInLockQueue()
/* --------------------------------------------------------------------------------- */
/* WE WILL CHECK IF THIS TRANSACTION IS ALREADY PLACED AT SOME SPOT IN THE PARALLEL */
/* SERIAL QUEUE WITHOUT ANY NEIGHBORS FROM OTHER TRANSACTION. IF SO WE WILL INSERT */
/* IT IN THAT PARALLEL QUEUE. */
/* --------------------------------------------------------------------------------- */
void Dbacc::placeSerialQueueRead(Signal* signal)
{
readWriteOpPtr.i = queOperPtr.p->nextSerialQue;
ptrCheckGuard(readWriteOpPtr, coprecsize, operationrec);
PSQR_LOOP:
jam();
if (readWriteOpPtr.p->nextSerialQue == RNIL) {
jam();
/* --------------------------------------------------------------------------------- */
/* THERE WAS NO PREVIOUS OPERATION IN THIS TRANSACTION WHICH WE COULD PUT IT */
/* IN THE PARALLEL QUEUE TOGETHER WITH. */
/* --------------------------------------------------------------------------------- */
checkOnlyReadEntry(signal);
return;
}//if
if (getNoParallelTransaction(readWriteOpPtr.p) == 1) {
jam();
/* --------------------------------------------------------------------------------- */
/* THERE WAS ONLY ONE TRANSACTION INVOLVED IN THE PARALLEL QUEUE. IF THIS IS OUR */
/* TRANSACTION WE CAN STILL GET HOLD OF THE LOCK. */
/* --------------------------------------------------------------------------------- */
if ((readWriteOpPtr.p->transId1 == operationRecPtr.p->transId1) &&
(readWriteOpPtr.p->transId2 == operationRecPtr.p->transId2)) {
jam();
/* --------------------------------------------------------------------------------- */
/* WE ARE PERFORMING A READ IN THE SAME TRANSACTION WHERE WE ALREADY */
/* PREVIOUSLY HAVE EXECUTED AN OPERATION. INSERT-DELETE, READ-UPDATE, READ-READ, */
/* UPDATE-UPDATE, UPDATE-DELETE, READ-DELETE, INSERT-READ, INSERT-UPDATE ARE ALLOWED */
/* COMBINATIONS. A NEW INSERT AFTER A DELETE IS NOT ALLOWED AND SUCH AN INSERT WILL */
/* GO TO THE SERIAL LOCK QUEUE WHICH IT WILL NOT LEAVE UNTIL A TIME-OUT AND THE */
/* TRANSACTION IS ABORTED. READS AND UPDATES AFTER DELETES IS ALSO NOT ALLOWED. */
/* --------------------------------------------------------------------------------- */
mlpqOperPtr = readWriteOpPtr;
moveLastParallelQueue(signal);
readWriteOpPtr = mlpqOperPtr;
operationRecPtr.p->prevParallelQue = readWriteOpPtr.i;
readWriteOpPtr.p->nextParallelQue = operationRecPtr.i;
operationRecPtr.p->localdata[0] = readWriteOpPtr.p->localdata[0];
operationRecPtr.p->localdata[1] = readWriteOpPtr.p->localdata[1];
switch (readWriteOpPtr.p->lockMode) {
case ZREADLOCK:
jam();
/*empty*/;
break;
default:
jam();
/* --------------------------------------------------------------------------------- */
/* IF THE TRANSACTION PREVIOUSLY SET A WRITE LOCK WE MUST ENSURE THAT ALL */
/* OPERATIONS IN THE PARALLEL QUEUE HAVE WRITE LOCK MODE TO AVOID STRANGE BUGS.*/
/* --------------------------------------------------------------------------------- */
operationRecPtr.p->lockMode = readWriteOpPtr.p->lockMode;
break;
}//switch
putOpInFragWaitQue(signal);
return;
}//if
}//if
readWriteOpPtr.i = readWriteOpPtr.p->nextSerialQue;
ptrCheckGuard(readWriteOpPtr, coprecsize, operationrec);
goto PSQR_LOOP;
}//Dbacc::placeSerialQueueRead()
/* --------------------------------------------------------------------------------- */
/* WE WILL CHECK IF THE LAST ENTRY IN THE SERIAL QUEUE CONTAINS ONLY READ */
/* OPERATIONS. IF SO WE WILL INSERT IT IN THAT PARALLEL QUEUE. OTHERWISE WE */
/* WILL PLACE IT AT THE END OF THE SERIAL QUEUE. */
/* --------------------------------------------------------------------------------- */
void Dbacc::checkOnlyReadEntry(Signal* signal)
{
switch (readWriteOpPtr.p->lockMode) {
case ZREADLOCK:
jam();
/* --------------------------------------------------------------------------------- */
/* SINCE THIS LAST QUEUE ONLY CONTAINS READ LOCKS WE CAN JOIN THE PARALLEL QUEUE AT */
/* THE END. */
/* --------------------------------------------------------------------------------- */
mlpqOperPtr = readWriteOpPtr;
moveLastParallelQueue(signal);
readWriteOpPtr = mlpqOperPtr;
operationRecPtr.p->prevParallelQue = readWriteOpPtr.i;
readWriteOpPtr.p->nextParallelQue = operationRecPtr.i;
operationRecPtr.p->localdata[0] = readWriteOpPtr.p->localdata[0];
operationRecPtr.p->localdata[1] = readWriteOpPtr.p->localdata[1];
break;
default:
jam(); /* PUT THE OPERATION RECORD IN THE SERIAL QUEUE */
readWriteOpPtr.p->nextSerialQue = operationRecPtr.i;
operationRecPtr.p->prevSerialQue = readWriteOpPtr.i;
break;
}//switch
putOpInFragWaitQue(signal);
}//Dbacc::checkOnlyReadEntry()
/* --------------------------------------------------------------------------------- */
/* GET_NO_PARALLEL_TRANSACTION */
/* --------------------------------------------------------------------------------- */
Uint32
Dbacc::getNoParallelTransaction(const Operationrec * op)
{
OperationrecPtr tmp;
tmp.i= op->nextParallelQue;
Uint32 transId[2] = { op->transId1, op->transId2 };
while (tmp.i != RNIL)
{
jam();
ptrCheckGuard(tmp, coprecsize, operationrec);
if (tmp.p->transId1 == transId[0] && tmp.p->transId2 == transId[1])
tmp.i = tmp.p->nextParallelQue;
else
return 2;
}
return 1;
}//Dbacc::getNoParallelTransaction()
void Dbacc::moveLastParallelQueue(Signal* signal)
{
while (mlpqOperPtr.p->nextParallelQue != RNIL) {
jam();
mlpqOperPtr.i = mlpqOperPtr.p->nextParallelQue;
ptrCheckGuard(mlpqOperPtr, coprecsize, operationrec);
}//if
}//Dbacc::moveLastParallelQueue()
void Dbacc::moveLastParallelQueueWrite(Signal* signal)
{
/* --------------------------------------------------------------------------------- */
/* ENSURE THAT ALL OPERATIONS HAVE LOCK MODE SET TO WRITE SINCE WE INSERT A */
/* WRITE LOCK INTO THE PARALLEL QUEUE. */
/* --------------------------------------------------------------------------------- */
while (mlpqOperPtr.p->nextParallelQue != RNIL) {
jam();
mlpqOperPtr.p->lockMode = operationRecPtr.p->lockMode;
mlpqOperPtr.i = mlpqOperPtr.p->nextParallelQue;
ptrCheckGuard(mlpqOperPtr, coprecsize, operationrec);
}//if
mlpqOperPtr.p->lockMode = operationRecPtr.p->lockMode;
}//Dbacc::moveLastParallelQueueWrite()
/* --------------------------------------------------------------------------------- */
/* PLACE_WRITE_IN_LOCK_QUEUE */
/* INPUT: OPERATION_REC_PTR OUR OPERATION POINTER */
/* QUE_OPER_PTR LOCK QUEUE OWNER OPERATION POINTER */
/* PWI_PAGEPTR PAGE POINTER OF ELEMENT */
/* TPWI_ELEMENTPTR ELEMENT POINTER OF ELEMENT */
/* OUTPUT TRESULT = */
/* ZPARALLEL_QUEUE OPERATION PLACED IN PARALLEL QUEUE */
/* OPERATION CAN PROCEED NOW. */
/* ZSERIAL_QUEUE OPERATION PLACED IN SERIAL QUEUE */
/* ERROR CODE OPERATION NEEDS ABORTING */
/* --------------------------------------------------------------------------------- */
Uint32 Dbacc::placeWriteInLockQueue(Signal* signal)
{
if (!((getNoParallelTransaction(queOperPtr.p) == 1) &&
(queOperPtr.p->transId1 == operationRecPtr.p->transId1) &&
(queOperPtr.p->transId2 == operationRecPtr.p->transId2))) {
jam();
placeSerialQueueWrite(signal);
return ZSERIAL_QUEUE;
}//if
/*
WE ARE PERFORMING AN READ EXCLUSIVE, INSERT, UPDATE OR DELETE IN THE SAME
TRANSACTION WHERE WE PREVIOUSLY HAVE EXECUTED AN OPERATION.
Read-All, Update-All, Insert-All and Delete-Insert are allowed
combinations.
Delete-Read, Delete-Update and Delete-Delete are not an allowed
combination and will result in tuple not found error.
*/
mlpqOperPtr = queOperPtr;
moveLastParallelQueueWrite(signal);
if (operationRecPtr.p->operation == ZINSERT &&
mlpqOperPtr.p->operation != ZDELETE){
jam();
return ZWRITE_ERROR;
}//if
if(operationRecPtr.p->operation == ZWRITE)
{
operationRecPtr.p->operation =
(mlpqOperPtr.p->operation == ZDELETE) ? ZINSERT : ZUPDATE;
}
operationRecPtr.p->localdata[0] = queOperPtr.p->localdata[0];
operationRecPtr.p->localdata[1] = queOperPtr.p->localdata[1];
operationRecPtr.p->prevParallelQue = mlpqOperPtr.i;
mlpqOperPtr.p->nextParallelQue = operationRecPtr.i;
return ZPARALLEL_QUEUE;
}//Dbacc::placeWriteInLockQueue()
/* --------------------------------------------------------------------------------- */
/* WE HAVE TO PLACE IT SOMEWHERE IN THE SERIAL QUEUE INSTEAD. */
/* --------------------------------------------------------------------------------- */
void Dbacc::placeSerialQueueWrite(Signal* signal)
{
readWriteOpPtr = queOperPtr;
PSQW_LOOP:
if (readWriteOpPtr.p->nextSerialQue == RNIL) {
jam();
/* --------------------------------------------------------------------------------- */
/* WE COULD NOT PUT IN ANY PARALLEL QUEUE. WE MUST PUT IT LAST IN THE SERIAL QUEUE. */
/* --------------------------------------------------------------------------------- */
readWriteOpPtr.p->nextSerialQue = operationRecPtr.i;
operationRecPtr.p->prevSerialQue = readWriteOpPtr.i;
putOpInFragWaitQue(signal);
return;
}//if
readWriteOpPtr.i = readWriteOpPtr.p->nextSerialQue;
ptrCheckGuard(readWriteOpPtr, coprecsize, operationrec);
if (getNoParallelTransaction(readWriteOpPtr.p) == 1) {
/* --------------------------------------------------------------------------------- */
/* THERE WAS ONLY ONE TRANSACTION INVOLVED IN THE PARALLEL QUEUE. IF THIS IS OUR */
/* TRANSACTION WE CAN STILL GET HOLD OF THE LOCK. */
/* --------------------------------------------------------------------------------- */
if ((readWriteOpPtr.p->transId1 == operationRecPtr.p->transId1) &&
(readWriteOpPtr.p->transId2 == operationRecPtr.p->transId2)) {
jam();
/* --------------------------------------------------------------------------------- */
/* WE ARE PERFORMING AN UPDATE OR DELETE IN THE SAME TRANSACTION WHERE WE ALREADY */
/* PREVIOUSLY HAVE EXECUTED AN OPERATION. INSERT-DELETE, READ-UPDATE, READ-READ, */
/* UPDATE-UPDATE, UPDATE-DELETE, READ-DELETE, INSERT-READ, INSERT-UPDATE ARE ALLOWED */
/* COMBINATIONS. A NEW INSERT AFTER A DELETE IS NOT ALLOWED AND SUCH AN INSERT WILL */
/* GO TO THE SERIAL LOCK QUEUE WHICH IT WILL NOT LEAVE UNTIL A TIME-OUT AND THE */
/* TRANSACTION IS ABORTED. READS AND UPDATES AFTER DELETES IS ALSO NOT ALLOWED. */
/* --------------------------------------------------------------------------------- */
mlpqOperPtr = readWriteOpPtr;
moveLastParallelQueueWrite(signal);
readWriteOpPtr = mlpqOperPtr;
operationRecPtr.p->prevParallelQue = readWriteOpPtr.i;
readWriteOpPtr.p->nextParallelQue = operationRecPtr.i;
operationRecPtr.p->localdata[0] = readWriteOpPtr.p->localdata[0];
operationRecPtr.p->localdata[1] = readWriteOpPtr.p->localdata[1];
putOpInFragWaitQue(signal);
return;
}//if
}//if
goto PSQW_LOOP;
}//Dbacc::placeSerialQueueWrite()
/* ------------------------------------------------------------------------- */
/* ACC KEYREQ END */
/* ------------------------------------------------------------------------- */
void Dbacc::acckeyref1Lab(Signal* signal, Uint32 result_code)
{
if (operationRecPtr.p->keyinfoPage != RNIL) {
jam();
rpPageptr.i = operationRecPtr.p->keyinfoPage;
ptrCheckGuard(rpPageptr, cpagesize, page8);
releasePage(signal);
operationRecPtr.p->keyinfoPage = RNIL;
}//if
operationRecPtr.p->transactionstate = WAIT_COMMIT_ABORT;
/* ************************<< */
/* ACCKEYREF */
/* ************************<< */
signal->theData[0] = cminusOne;
signal->theData[1] = result_code;
return;
}//Dbacc::acckeyref1Lab()
/* ******************--------------------------------------------------------------- */
/* ACCMINUPDATE UPDATE LOCAL KEY REQ */
/* DESCRIPTION: UPDATES LOCAL KEY OF AN ELEMENTS IN THE HASH TABLE */
/* THIS SIGNAL IS WAITED AFTER ANY INSERT REQ */
/* ENTER ACCMINUPDATE WITH SENDER: LQH, LEVEL B */
/* OPERATION_REC_PTR, OPERATION RECORD PTR */
/* CLOCALKEY(0), LOCAL KEY 1 */
/* CLOCALKEY(1) LOCAL KEY 2 */
/* ******************--------------------------------------------------------------- */
void Dbacc::execACCMINUPDATE(Signal* signal)
{
Page8Ptr ulkPageidptr;
Uint32 tulkLocalPtr;
Uint32 tlocalkey1, tlocalkey2;
Uint32 TlogStart;
jamEntry();
operationRecPtr.i = signal->theData[0];
tlocalkey1 = signal->theData[1];
tlocalkey2 = signal->theData[2];
ptrCheckGuard(operationRecPtr, coprecsize, operationrec);
if (operationRecPtr.p->transactionstate == ACTIVE) {
fragrecptr.i = operationRecPtr.p->fragptr;
ulkPageidptr.i = operationRecPtr.p->elementPage;
tulkLocalPtr = operationRecPtr.p->elementPointer + operationRecPtr.p->elementIsforward;
ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
ptrCheckGuard(ulkPageidptr, cpagesize, page8);
if (fragrecptr.p->createLcp == ZTRUE) {
//----------------------------------------------------------
// To avoid undo log the element header we take care to only
// undo log the local key part.
//----------------------------------------------------------
if (operationRecPtr.p->elementIsforward == 1) {
jam();
TlogStart = tulkLocalPtr;
} else {
jam();
TlogStart = tulkLocalPtr - fragrecptr.p->localkeylen + 1;
}//if
datapageptr.p = ulkPageidptr.p;
cundoinfolength = fragrecptr.p->localkeylen;
cundoElemIndex = TlogStart;
undoWritingProcess(signal);
}//if
dbgWord32(ulkPageidptr, tulkLocalPtr, tlocalkey1);
arrGuard(tulkLocalPtr, 2048);
ulkPageidptr.p->word32[tulkLocalPtr] = tlocalkey1;
operationRecPtr.p->localdata[0] = tlocalkey1;
if (fragrecptr.p->localkeylen == 1) {
return;
} else if (fragrecptr.p->localkeylen == 2) {
jam();
tulkLocalPtr = tulkLocalPtr + operationRecPtr.p->elementIsforward;
operationRecPtr.p->localdata[1] = tlocalkey2;
dbgWord32(ulkPageidptr, tulkLocalPtr, tlocalkey2);
arrGuard(tulkLocalPtr, 2048);
ulkPageidptr.p->word32[tulkLocalPtr] = tlocalkey2;
return;
} else {
jam();
}//if
}//if
ndbrequire(false);
}//Dbacc::execACCMINUPDATE()
/* ******************--------------------------------------------------------------- */
/* ACC_COMMITREQ COMMIT TRANSACTION */
/* SENDER: LQH, LEVEL B */
/* INPUT: OPERATION_REC_PTR , */
/* ******************--------------------------------------------------------------- */
void Dbacc::execACC_COMMITREQ(Signal* signal)
{
Uint8 Toperation;
jamEntry();
operationRecPtr.i = signal->theData[0];
ptrCheckGuard(operationRecPtr, coprecsize, operationrec);
ndbrequire(operationRecPtr.p->transactionstate == ACTIVE);
fragrecptr.i = operationRecPtr.p->fragptr;
ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
commitOperation(signal);
Toperation = operationRecPtr.p->operation;
operationRecPtr.p->transactionstate = IDLE;
operationRecPtr.p->operation = ZUNDEFINED_OP;
if(Toperation != ZREAD){
rootfragrecptr.p->m_commit_count++;
if (Toperation != ZINSERT) {
if (Toperation != ZDELETE) {
return;
} else {
jam();
rootfragrecptr.i = fragrecptr.p->myroot;
ptrCheckGuard(rootfragrecptr, crootfragmentsize, rootfragmentrec);
rootfragrecptr.p->noOfElements--;
fragrecptr.p->slack += operationRecPtr.p->insertDeleteLen;
if (fragrecptr.p->slack > fragrecptr.p->slackCheck) {
/* TIME FOR JOIN BUCKETS PROCESS */
if (fragrecptr.p->expandCounter > 0) {
if (fragrecptr.p->expandFlag < 2) {
jam();
signal->theData[0] = fragrecptr.i;
signal->theData[1] = fragrecptr.p->p;
signal->theData[2] = fragrecptr.p->maxp;
signal->theData[3] = fragrecptr.p->expandFlag;
fragrecptr.p->expandFlag = 2;
sendSignal(cownBlockref, GSN_SHRINKCHECK2, signal, 4, JBB);
}//if
}//if
}//if
}//if
} else {
jam(); /* EXPAND PROCESS HANDLING */
rootfragrecptr.i = fragrecptr.p->myroot;
ptrCheckGuard(rootfragrecptr, crootfragmentsize, rootfragmentrec);
rootfragrecptr.p->noOfElements++;
fragrecptr.p->slack -= operationRecPtr.p->insertDeleteLen;
if (fragrecptr.p->slack >= (1u << 31)) {
/* IT MEANS THAT IF SLACK < ZERO */
if (fragrecptr.p->expandFlag == 0) {
jam();
fragrecptr.p->expandFlag = 2;
signal->theData[0] = fragrecptr.i;
signal->theData[1] = fragrecptr.p->p;
signal->theData[2] = fragrecptr.p->maxp;
sendSignal(cownBlockref, GSN_EXPANDCHECK2, signal, 3, JBB);
}//if
}//if
}//if
}
return;
}//Dbacc::execACC_COMMITREQ()
/* ******************--------------------------------------------------------------- */
/* ACC ABORT REQ ABORT ALL OPERATION OF THE TRANSACTION */
/* ******************------------------------------+ */
/* SENDER: LQH, LEVEL B */
/* ******************--------------------------------------------------------------- */
/* ACC ABORT REQ ABORT TRANSACTION */
/* ******************------------------------------+ */
/* SENDER: LQH, LEVEL B */
void Dbacc::execACC_ABORTREQ(Signal* signal)
{
jamEntry();
accAbortReqLab(signal, true);
}//Dbacc::execACC_ABORTREQ()
void Dbacc::accAbortReqLab(Signal* signal, bool sendConf)
{
operationRecPtr.i = signal->theData[0];
ptrCheckGuard(operationRecPtr, coprecsize, operationrec);
tresult = 0; /* ZFALSE */
if ((operationRecPtr.p->transactionstate == ACTIVE) ||
(operationRecPtr.p->transactionstate == WAIT_COMMIT_ABORT)) {
jam();
fragrecptr.i = operationRecPtr.p->fragptr;
ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
operationRecPtr.p->transactionstate = ABORT;
abortOperation(signal);
} else {
ndbrequire(operationRecPtr.p->transactionstate == IDLE);
jam();
}//if
operationRecPtr.p->transactionstate = IDLE;
operationRecPtr.p->operation = ZUNDEFINED_OP;
if (! sendConf)
return;
signal->theData[0] = operationRecPtr.p->userptr;
sendSignal(operationRecPtr.p->userblockref, GSN_ACC_ABORTCONF, signal, 1, JBB);
return;
}//Dbacc::accAbortReqLab()
/*
* Lock or unlock tuple.
*/
void Dbacc::execACC_LOCKREQ(Signal* signal)
{
jamEntry();
AccLockReq* sig = (AccLockReq*)signal->getDataPtrSend();
AccLockReq reqCopy = *sig;
AccLockReq* const req = &reqCopy;
Uint32 lockOp = (req->requestInfo & 0xFF);
if (lockOp == AccLockReq::LockShared ||
lockOp == AccLockReq::LockExclusive) {
jam();
// find table
tabptr.i = req->tableId;
ptrCheckGuard(tabptr, ctablesize, tabrec);
// find fragment (TUX will know it)
if (req->fragPtrI == RNIL) {
for (Uint32 i = 0; i < MAX_FRAG_PER_NODE; i++) {
jam();
if (tabptr.p->fragptrholder[i] != RNIL) {
rootfragrecptr.i = tabptr.p->fragptrholder[i];
ptrCheckGuard(rootfragrecptr, crootfragmentsize, rootfragmentrec);
if (rootfragrecptr.p->fragmentid[0] == req->fragId) {
jam();
req->fragPtrI = rootfragrecptr.p->fragmentptr[0];
break;
}
if (rootfragrecptr.p->fragmentid[1] == req->fragId) {
jam();
req->fragPtrI = rootfragrecptr.p->fragmentptr[1];
break;
}
}
}
}
fragrecptr.i = req->fragPtrI;
ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
ndbrequire(req->fragId == fragrecptr.p->myfid);
// caller must be explicit here
ndbrequire(req->accOpPtr == RNIL);
// seize operation to hold the lock
if (cfreeopRec != RNIL) {
jam();
seizeOpRec(signal);
// init as in ACCSEIZEREQ
operationRecPtr.p->userptr = req->userPtr;
operationRecPtr.p->userblockref = req->userRef;
operationRecPtr.p->operation = ZUNDEFINED_OP;
operationRecPtr.p->transactionstate = IDLE;
// do read with lock via ACCKEYREQ
Uint32 lockMode = (lockOp == AccLockReq::LockShared) ? 0 : 1;
Uint32 opCode = ZSCAN_OP;
signal->theData[0] = operationRecPtr.i;
signal->theData[1] = fragrecptr.i;
signal->theData[2] = opCode | (lockMode << 4) | (1u << 31);
signal->theData[3] = req->hashValue;
signal->theData[4] = 1; // fake primKeyLen
signal->theData[5] = req->transId1;
signal->theData[6] = req->transId2;
signal->theData[7] = req->tupAddr;
EXECUTE_DIRECT(DBACC, GSN_ACCKEYREQ, signal, 8);
// translate the result
if (signal->theData[0] < RNIL) {
jam();
req->returnCode = AccLockReq::Success;
req->accOpPtr = operationRecPtr.i;
} else if (signal->theData[0] == RNIL) {
jam();
req->returnCode = AccLockReq::IsBlocked;
req->accOpPtr = operationRecPtr.i;
} else {
ndbrequire(signal->theData[0] == (UintR)-1);
releaseOpRec(signal);
req->returnCode = AccLockReq::Refused;
req->accOpPtr = RNIL;
}
} else {
jam();
req->returnCode = AccLockReq::NoFreeOp;
}
*sig = *req;
return;
}
operationRecPtr.i = req->accOpPtr;
ptrCheckGuard(operationRecPtr, coprecsize, operationrec);
fragrecptr.i = operationRecPtr.p->fragptr;
ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
if (fragrecptr.p->keyLength == 0 &&
// should test some state variable
operationRecPtr.p->elementPage != RNIL) {
jam();
// re-compute long key vars
Page8Ptr tPageptr;
tPageptr.i = operationRecPtr.p->elementPage;
ptrCheckGuard(tPageptr, cpagesize, page8);
Uint32 tKeyptr =
operationRecPtr.p->elementPointer +
operationRecPtr.p->elementIsforward *
(ZELEM_HEAD_SIZE + fragrecptr.p->localkeylen);
tslcPageIndex = tPageptr.p->word32[tKeyptr] & 0x3ff;
tslcPagedir = tPageptr.p->word32[tKeyptr] >> 10;
searchLongKey(signal, false);
}
if (lockOp == AccLockReq::Unlock) {
jam();
// do unlock via ACC_COMMITREQ (immediate)
signal->theData[0] = req->accOpPtr;
EXECUTE_DIRECT(DBACC, GSN_ACC_COMMITREQ, signal, 1);
releaseOpRec(signal);
req->returnCode = AccLockReq::Success;
*sig = *req;
return;
}
if (lockOp == AccLockReq::Abort) {
jam();
// do abort via ACC_ABORTREQ (immediate)
signal->theData[0] = req->accOpPtr;
accAbortReqLab(signal, false);
releaseOpRec(signal);
req->returnCode = AccLockReq::Success;
*sig = *req;
return;
}
if (lockOp == AccLockReq::AbortWithConf) {
jam();
// do abort via ACC_ABORTREQ (with conf signal)
signal->theData[0] = req->accOpPtr;
accAbortReqLab(signal, true);
releaseOpRec(signal);
req->returnCode = AccLockReq::Success;
*sig = *req;
return;
}
ndbrequire(false);
}
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* */
/* END OF EXECUTE OPERATION MODULE */
/* */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* */
/* MODULE: INSERT */
/* THE FOLLOWING SUBROUTINES ARE ONLY USED BY INSERT_ELEMENT. THIS */
/* ROUTINE IS THE SOLE INTERFACE TO INSERT ELEMENTS INTO THE INDEX. */
/* CURRENT USERS ARE INSERT REQUESTS, EXPAND CONTAINER AND SHRINK */
/* CONTAINER. */
/* */
/* THE FOLLOWING SUBROUTINES ARE INCLUDED IN THIS MODULE: */
/* INSERT_ELEMENT */
/* INSERT_CONTAINER */
/* ADDNEWCONTAINER */
/* GETFREELIST */
/* INCREASELISTCONT */
/* SEIZE_LEFTLIST */
/* SEIZE_RIGHTLIST */
/* */
/* THESE ROUTINES ARE ONLY USED BY THIS MODULE AND BY NO ONE ELSE. */
/* ALSO THE ROUTINES MAKE NO USE OF ROUTINES IN OTHER MODULES. */
/* TAKE_REC_OUT_OF_FREE_OVERPAGE AND RELEASE_OVERFLOW_REC ARE */
/* EXCEPTIONS TO THIS RULE. */
/* */
/* THE ONLY SHORT-LIVED VARIABLES USED IN OTHER PARTS OF THE BLOCK ARE */
/* THOSE DEFINED AS INPUT AND OUTPUT IN INSERT_ELEMENT */
/* SHORT-LIVED VARIABLES INCLUDE TEMPORARY VARIABLES, COMMON VARIABLES */
/* AND POINTER VARIABLES. */
/* THE ONLY EXCEPTION TO THIS RULE IS FRAGRECPTR WHICH POINTS TO THE */
/* FRAGMENT RECORD. THIS IS MORE LESS STATIC ALWAYS DURING A SIGNAL */
/* EXECUTION. */
/* */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* --------------------------------------------------------------------------------- */
/* INSERT_ELEMENT */
/* INPUT: */
/* IDR_PAGEPTR (POINTER TO THE ACTIVE PAGE REC) */
/* TIDR_PAGEINDEX (INDEX OF THE CONTAINER) */
/* TIDR_FORWARD (DIRECTION FORWARD OR BACKWARD) */
/* TIDR_ELEMHEAD (HEADER OF ELEMENT TO BE INSERTED */
/* CIDR_KEYS(ARRAY OF TUPLE KEYS) */
/* CLOCALKEY(ARRAY OF LOCAL KEYS). */
/* FRAGRECPTR */
/* IDR_OPERATION_REC_PTR */
/* TIDR_KEY_LEN */
/* */
/* OUTPUT: */
/* TIDR_PAGEINDEX (PAGE INDEX OF INSERTED ELEMENT) */
/* IDR_PAGEPTR (PAGE POINTER OF INSERTED ELEMENT) */
/* TIDR_FORWARD (CONTAINER DIRECTION OF INSERTED ELEMENT) */
/* NONE */
/* --------------------------------------------------------------------------------- */
void Dbacc::insertElement(Signal* signal)
{
DirRangePtr inrOverflowrangeptr;
DirectoryarrayPtr inrOverflowDirptr;
OverflowRecordPtr inrOverflowRecPtr;
Page8Ptr inrNewPageptr;
Uint32 tinrNextSamePage;
Uint32 tinrTmp;
do {
insertContainer(signal);
if (tidrResult != ZFALSE) {
jam();
return;
/* INSERTION IS DONE, OR */
/* AN ERROR IS DETECTED */
}//if
if (((tidrContainerhead >> 7) & 0x3) != 0) {
tinrNextSamePage = (tidrContainerhead >> 9) & 0x1; /* CHECK BIT FOR CHECKING WHERE */
/* THE NEXT CONTAINER IS IN THE SAME PAGE */
tidrPageindex = tidrContainerhead & 0x7f; /* NEXT CONTAINER PAGE INDEX 7 BITS */
if (((tidrContainerhead >> 7) & 3) == ZLEFT) {
jam();
tidrForward = ZTRUE;
} else if (((tidrContainerhead >> 7) & 3) == ZRIGHT) {
jam();
tidrForward = cminusOne;
} else {
ndbrequire(false);
return;
}//if
if (tinrNextSamePage == ZFALSE) {
jam(); /* NEXT CONTAINER IS IN AN OVERFLOW PAGE */
tinrTmp = idrPageptr.p->word32[tidrContainerptr + 1];
inrOverflowrangeptr.i = fragrecptr.p->overflowdir;
ptrCheckGuard(inrOverflowrangeptr, cdirrangesize, dirRange);
arrGuard((tinrTmp >> 8), 256);
inrOverflowDirptr.i = inrOverflowrangeptr.p->dirArray[tinrTmp >> 8];
ptrCheckGuard(inrOverflowDirptr, cdirarraysize, directoryarray);
idrPageptr.i = inrOverflowDirptr.p->pagep[tinrTmp & 0xff];
ptrCheckGuard(idrPageptr, cpagesize, page8);
}//if
ndbrequire(tidrPageindex < ZEMPTYLIST);
} else {
break;
}//if
} while (1);
gflPageptr.p = idrPageptr.p;
getfreelist(signal);
if (tgflPageindex == ZEMPTYLIST) {
jam();
/* NO FREE BUFFER IS FOUND */
if (fragrecptr.p->firstOverflowRec == RNIL) {
jam();
allocOverflowPage(signal);
ndbrequire(tresult <= ZLIMIT_OF_ERROR);
}//if
inrOverflowRecPtr.i = fragrecptr.p->firstOverflowRec;
ptrCheckGuard(inrOverflowRecPtr, coverflowrecsize, overflowRecord);
inrNewPageptr.i = inrOverflowRecPtr.p->overpage;
ptrCheckGuard(inrNewPageptr, cpagesize, page8);
gflPageptr.p = inrNewPageptr.p;
getfreelist(signal);
ndbrequire(tgflPageindex != ZEMPTYLIST);
tancNext = 0;
} else {
jam();
inrNewPageptr = idrPageptr;
tancNext = 1;
}//if
tslUpdateHeader = ZTRUE;
tslPageindex = tgflPageindex;
slPageptr.p = inrNewPageptr.p;
if (tgflBufType == ZLEFT) {
seizeLeftlist(signal);
tidrForward = ZTRUE;
} else {
seizeRightlist(signal);
tidrForward = cminusOne;
}//if
tancPageindex = tgflPageindex;
tancPageid = inrNewPageptr.p->word32[ZPOS_PAGE_ID];
tancBufType = tgflBufType;
tancContainerptr = tidrContainerptr;
ancPageptr.p = idrPageptr.p;
addnewcontainer(signal);
idrPageptr = inrNewPageptr;
tidrPageindex = tgflPageindex;
insertContainer(signal);
ndbrequire(tidrResult == ZTRUE);
}//Dbacc::insertElement()
/* --------------------------------------------------------------------------------- */
/* INSERT_CONTAINER */
/* INPUT: */
/* IDR_PAGEPTR (POINTER TO THE ACTIVE PAGE REC) */
/* TIDR_PAGEINDEX (INDEX OF THE CONTAINER) */
/* TIDR_FORWARD (DIRECTION FORWARD OR BACKWARD) */
/* TIDR_ELEMHEAD (HEADER OF ELEMENT TO BE INSERTED */
/* CKEYS(ARRAY OF TUPLE KEYS) */
/* CLOCALKEY(ARRAY 0F LOCAL KEYS). */
/* TIDR_KEY_LEN */
/* FRAGRECPTR */
/* IDR_OPERATION_REC_PTR */
/* OUTPUT: */
/* TIDR_RESULT (ZTRUE FOR SUCCESS AND ZFALSE OTHERWISE) */
/* TIDR_CONTAINERHEAD (HEADER OF CONTAINER) */
/* TIDR_CONTAINERPTR (POINTER TO CONTAINER HEADER) */
/* */
/* DESCRIPTION: */
/* THE FREE AREA OF THE CONTAINER WILL BE CALCULATED. IF IT IS */
/* LARGER THAN OR EQUAL THE ELEMENT LENGTH. THE ELEMENT WILL BE */
/* INSERT IN THE CONTAINER AND CONTAINER HEAD WILL BE UPDATED. */
/* THIS ROUTINE ALWAYS DEALS WITH ONLY ONE CONTAINER AND DO NEVER */
/* START ANYTHING OUTSIDE OF THIS CONTAINER. */
/* */
/* SHORT FORM: IDR */
/* --------------------------------------------------------------------------------- */
void Dbacc::insertContainer(Signal* signal)
{
Uint32 tidrContainerlen;
Uint32 tidrConfreelen;
Uint32 tidrNextSide;
Uint32 tidrNextConLen;
Uint32 tidrIndex;
Uint32 tidrInputIndex;