Skip to content

M
mariadb
Commit 9a913689 authored by tomas@poseidon.ndb.mysql.com's avatar tomas@poseidon.ndb.mysql.com
Browse files
Options

improved the ndb redo log reader

parent 298af8cc
Show whitespace changes
Inline Side-by-side
Showing with 105 additions and 164 deletions
ndb/src/kernel/blocks/dblqh/redoLogReader/records.cpp
View file @ 9a913689
...@@ -239,6 +239,17 @@ bool PageHeader::check() { ...@@ -239,6 +239,17 @@ bool PageHeader::check() {
return true; return true;
} }
bool PageHeader::lastPage()
{
return m_next_page == 0xffffff00;
}
Uint32 PageHeader::lastWord()
{
return m_current_page_index;
}
NdbOut& operator<<(NdbOut& no, const PageHeader& ph) { NdbOut& operator<<(NdbOut& no, const PageHeader& ph) {
no << "------------PAGE HEADER------------------------" << endl << endl; no << "------------PAGE HEADER------------------------" << endl << endl;
ndbout_c("%-30s%-12s%-12s\n", "", "Decimal", "Hex"); ndbout_c("%-30s%-12s%-12s\n", "", "Decimal", "Hex");
......
ndb/src/kernel/blocks/dblqh/redoLogReader/records.hpp
View file @ 9a913689
...@@ -132,6 +132,8 @@ class PageHeader { ...@@ -132,6 +132,8 @@ class PageHeader {
public: public:
bool check(); bool check();
Uint32 getLogRecordSize(); Uint32 getLogRecordSize();
bool lastPage();
Uint32 lastWord();
protected: protected:
Uint32 m_checksum; Uint32 m_checksum;
Uint32 m_lap; Uint32 m_lap;
......
ndb/src/kernel/blocks/dblqh/redoLogReader/redoLogFileReader.cpp
View file @ 9a913689
...@@ -35,7 +35,6 @@ ...@@ -35,7 +35,6 @@
#define FROM_BEGINNING 0 #define FROM_BEGINNING 0
void usage(const char * prg); void usage(const char * prg);
Uint32 readRecordOverPageBoundary (Uint32 *, Uint32 , Uint32 , Uint32);
Uint32 readFromFile(FILE * f, Uint32 *toPtr, Uint32 sizeInWords); Uint32 readFromFile(FILE * f, Uint32 *toPtr, Uint32 sizeInWords);
void readArguments(int argc, const char** argv); void readArguments(int argc, const char** argv);
void doExit(); void doExit();
...@@ -54,8 +53,8 @@ Uint32 startAtPageIndex = 0; ...@@ -54,8 +53,8 @@ Uint32 startAtPageIndex = 0;
Uint32 *redoLogPage; Uint32 *redoLogPage;
NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read a redo log file", 16384) { NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read a redo log file", 16384) {
Uint32 pageIndex = 0; int wordIndex = 0;
Uint32 oldPageIndex = 0; int oldWordIndex = 0;
Uint32 recordType = 1234567890; Uint32 recordType = 1234567890;
PageHeader *thePageHeader; PageHeader *thePageHeader;
...@@ -83,25 +82,23 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read ...@@ -83,25 +82,23 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read
} }
redoLogPage = new Uint32[PAGESIZE*NO_PAGES_IN_MBYTE]; redoLogPage = new Uint32[PAGESIZE*NO_PAGES_IN_MBYTE];
Uint32 words_from_previous_page = 0;
// Loop for every mbyte. // Loop for every mbyte.
for (Uint32 j = startAtMbyte; j < NO_MBYTE_IN_FILE; j++) { bool lastPage = false;
for (Uint32 j = startAtMbyte; j < NO_MBYTE_IN_FILE && !lastPage; j++) {
readFromFile(f, redoLogPage, PAGESIZE*NO_PAGES_IN_MBYTE); readFromFile(f, redoLogPage, PAGESIZE*NO_PAGES_IN_MBYTE);
if (firstLap) { words_from_previous_page = 0;
pageIndex = startAtPageIndex;
firstLap = false;
} else
pageIndex = 0;
// Loop for every page. // Loop for every page.
for (int i = startAtPage; i < NO_PAGES_IN_MBYTE; i++) { for (int i = 0; i < NO_PAGES_IN_MBYTE; i++) {
wordIndex = 0;
if (pageIndex == 0) {
thePageHeader = (PageHeader *) &redoLogPage[i*PAGESIZE]; thePageHeader = (PageHeader *) &redoLogPage[i*PAGESIZE];
// Print out mbyte number, page number and page index. // Print out mbyte number, page number and page index.
ndbout << j << ":" << i << ":" << pageIndex << endl ndbout << j << ":" << i << ":" << wordIndex << endl
<< " " << j*32 + i << ":" << pageIndex << " "; << " " << j*32 + i << ":" << wordIndex << " ";
if (thePrintFlag) ndbout << (*thePageHeader); if (thePrintFlag) ndbout << (*thePageHeader);
if (theCheckFlag) { if (theCheckFlag) {
if(!thePageHeader->check()) { if(!thePageHeader->check()) {
...@@ -122,8 +119,9 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read ...@@ -122,8 +119,9 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read
ndbout << "expected checksum: " << checkSum << endl; ndbout << "expected checksum: " << checkSum << endl;
} }
pageIndex += thePageHeader->getLogRecordSize();
} lastPage = i != 0 && thePageHeader->lastPage();
Uint32 lastWord = thePageHeader->lastWord();
if (onlyMbyteHeaders) { if (onlyMbyteHeaders) {
// Show only the first page header in every mbyte of the file. // Show only the first page header in every mbyte of the file.
...@@ -132,18 +130,40 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read ...@@ -132,18 +130,40 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read
if (onlyPageHeaders) { if (onlyPageHeaders) {
// Show only page headers. Continue with the next page in this for loop. // Show only page headers. Continue with the next page in this for loop.
pageIndex = 0;
continue; continue;
} }
wordIndex = thePageHeader->getLogRecordSize() - words_from_previous_page;
Uint32 *redoLogPagePos = redoLogPage + i*PAGESIZE;
if (words_from_previous_page)
{
memmove(redoLogPagePos + wordIndex ,
redoLogPagePos - words_from_previous_page,
words_from_previous_page*4);
}
do { do {
// Print out mbyte number, page number and page index. if (words_from_previous_page)
ndbout << j << ":" << i << ":" << pageIndex << endl {
<< " " << j*32 + i << ":" << pageIndex << " "; // Print out mbyte number, page number and word index.
recordType = redoLogPage[i*PAGESIZE + pageIndex]; ndbout << j << ":" << i-1 << ":" << PAGESIZE-words_from_previous_page << endl
<< j << ":" << i << ":" << wordIndex+words_from_previous_page << endl
<< " " << j*32 + i-1 << ":" << PAGESIZE-words_from_previous_page << " ";
words_from_previous_page = 0;
}
else
{
// Print out mbyte number, page number and word index.
ndbout << j << ":" << i << ":" << wordIndex << endl
<< " " << j*32 + i << ":" << wordIndex << " ";
}
redoLogPagePos = redoLogPage + i*PAGESIZE + wordIndex;
oldWordIndex = wordIndex;
recordType = *redoLogPagePos;
switch(recordType) { switch(recordType) {
case ZFD_TYPE: case ZFD_TYPE:
fdRecord = (FileDescriptor *) &redoLogPage[i*PAGESIZE + pageIndex]; fdRecord = (FileDescriptor *) redoLogPagePos;
if (thePrintFlag) ndbout << (*fdRecord); if (thePrintFlag) ndbout << (*fdRecord);
if (theCheckFlag) { if (theCheckFlag) {
if(!fdRecord->check()) { if(!fdRecord->check()) {
...@@ -155,13 +175,13 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read ...@@ -155,13 +175,13 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read
delete [] redoLogPage; delete [] redoLogPage;
exit(RETURN_OK); exit(RETURN_OK);
} }
pageIndex += fdRecord->getLogRecordSize(); wordIndex += fdRecord->getLogRecordSize();
break; break;
case ZNEXT_LOG_RECORD_TYPE: case ZNEXT_LOG_RECORD_TYPE:
nlRecord = (NextLogRecord *) (&redoLogPage[i*PAGESIZE] + pageIndex); nlRecord = (NextLogRecord *) redoLogPagePos;
pageIndex += nlRecord->getLogRecordSize(pageIndex); wordIndex += nlRecord->getLogRecordSize(wordIndex);
if (pageIndex <= PAGESIZE) { if (wordIndex <= PAGESIZE) {
if (thePrintFlag) ndbout << (*nlRecord); if (thePrintFlag) ndbout << (*nlRecord);
if (theCheckFlag) { if (theCheckFlag) {
if(!nlRecord->check()) { if(!nlRecord->check()) {
...@@ -173,9 +193,9 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read ...@@ -173,9 +193,9 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read
break; break;
case ZCOMPLETED_GCI_TYPE: case ZCOMPLETED_GCI_TYPE:
cGCIrecord = (CompletedGCIRecord *) &redoLogPage[i*PAGESIZE + pageIndex]; cGCIrecord = (CompletedGCIRecord *) redoLogPagePos;
pageIndex += cGCIrecord->getLogRecordSize(); wordIndex += cGCIrecord->getLogRecordSize();
if (pageIndex <= PAGESIZE) { if (wordIndex <= PAGESIZE) {
if (thePrintFlag) ndbout << (*cGCIrecord); if (thePrintFlag) ndbout << (*cGCIrecord);
if (theCheckFlag) { if (theCheckFlag) {
if(!cGCIrecord->check()) { if(!cGCIrecord->check()) {
...@@ -187,9 +207,9 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read ...@@ -187,9 +207,9 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read
break; break;
case ZPREP_OP_TYPE: case ZPREP_OP_TYPE:
poRecord = (PrepareOperationRecord *) &redoLogPage[i*PAGESIZE + pageIndex]; poRecord = (PrepareOperationRecord *) redoLogPagePos;
pageIndex += poRecord->getLogRecordSize(); wordIndex += poRecord->getLogRecordSize();
if (pageIndex <= PAGESIZE) { if (wordIndex <= PAGESIZE) {
if (thePrintFlag) ndbout << (*poRecord); if (thePrintFlag) ndbout << (*poRecord);
if (theCheckFlag) { if (theCheckFlag) {
if(!poRecord->check()) { if(!poRecord->check()) {
...@@ -198,15 +218,12 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read ...@@ -198,15 +218,12 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read
} }
} }
} }
else {
oldPageIndex = pageIndex - poRecord->getLogRecordSize();
}
break; break;
case ZCOMMIT_TYPE: case ZCOMMIT_TYPE:
ctRecord = (CommitTransactionRecord *) &redoLogPage[i*PAGESIZE + pageIndex]; ctRecord = (CommitTransactionRecord *) redoLogPagePos;
pageIndex += ctRecord->getLogRecordSize(); wordIndex += ctRecord->getLogRecordSize();
if (pageIndex <= PAGESIZE) { if (wordIndex <= PAGESIZE) {
if (thePrintFlag) ndbout << (*ctRecord); if (thePrintFlag) ndbout << (*ctRecord);
if (theCheckFlag) { if (theCheckFlag) {
if(!ctRecord->check()) { if(!ctRecord->check()) {
...@@ -215,15 +232,12 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read ...@@ -215,15 +232,12 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read
} }
} }
} }
else {
oldPageIndex = pageIndex - ctRecord->getLogRecordSize();
}
break; break;
case ZINVALID_COMMIT_TYPE: case ZINVALID_COMMIT_TYPE:
ictRecord = (InvalidCommitTransactionRecord *) &redoLogPage[i*PAGESIZE + pageIndex]; ictRecord = (InvalidCommitTransactionRecord *) redoLogPagePos;
pageIndex += ictRecord->getLogRecordSize(); wordIndex += ictRecord->getLogRecordSize();
if (pageIndex <= PAGESIZE) { if (wordIndex <= PAGESIZE) {
if (thePrintFlag) ndbout << (*ictRecord); if (thePrintFlag) ndbout << (*ictRecord);
if (theCheckFlag) { if (theCheckFlag) {
if(!ictRecord->check()) { if(!ictRecord->check()) {
...@@ -232,21 +246,18 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read ...@@ -232,21 +246,18 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read
} }
} }
} }
else {
oldPageIndex = pageIndex - ictRecord->getLogRecordSize();
}
break; break;
case ZNEXT_MBYTE_TYPE: case ZNEXT_MBYTE_TYPE:
nmRecord = (NextMbyteRecord *) &redoLogPage[i*PAGESIZE + pageIndex]; nmRecord = (NextMbyteRecord *) redoLogPagePos;
if (thePrintFlag) ndbout << (*nmRecord); if (thePrintFlag) ndbout << (*nmRecord);
i = NO_PAGES_IN_MBYTE; i = NO_PAGES_IN_MBYTE;
break; break;
case ZABORT_TYPE: case ZABORT_TYPE:
atRecord = (AbortTransactionRecord *) &redoLogPage[i*PAGESIZE + pageIndex]; atRecord = (AbortTransactionRecord *) redoLogPagePos;
pageIndex += atRecord->getLogRecordSize(); wordIndex += atRecord->getLogRecordSize();
if (pageIndex <= PAGESIZE) { if (wordIndex <= PAGESIZE) {
if (thePrintFlag) ndbout << (*atRecord); if (thePrintFlag) ndbout << (*atRecord);
if (theCheckFlag) { if (theCheckFlag) {
if(!atRecord->check()) { if(!atRecord->check()) {
...@@ -266,7 +277,7 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read ...@@ -266,7 +277,7 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read
ndbout << " ------ERROR: UNKNOWN RECORD TYPE------" << endl; ndbout << " ------ERROR: UNKNOWN RECORD TYPE------" << endl;
// Print out remaining data in this page // Print out remaining data in this page
for (int j = pageIndex; j < PAGESIZE; j++){ for (int j = wordIndex; j < PAGESIZE; j++){
Uint32 unknown = redoLogPage[i*PAGESIZE + j]; Uint32 unknown = redoLogPage[i*PAGESIZE + j];
ndbout_c("%-30d%-12u%-12x", j, unknown, unknown); ndbout_c("%-30d%-12u%-12x", j, unknown, unknown);
...@@ -274,14 +285,18 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read ...@@ -274,14 +285,18 @@ NDB_COMMAND(redoLogFileReader, "redoLogFileReader", "redoLogFileReader", "Read
doExit(); doExit();
} }
} while(pageIndex < PAGESIZE && i < NO_PAGES_IN_MBYTE); } while(wordIndex < lastWord && i < NO_PAGES_IN_MBYTE);
if (lastPage)
break;
if (pageIndex > PAGESIZE) { if (wordIndex > PAGESIZE) {
// The last record overlapped page boundary. Must redo that record. words_from_previous_page = PAGESIZE - oldWordIndex;
pageIndex = readRecordOverPageBoundary(&redoLogPage[i*PAGESIZE], ndbout << " ----------- Record continues on next page -----------" << endl;
pageIndex, oldPageIndex, recordType);
} else { } else {
pageIndex = 0; wordIndex = 0;
words_from_previous_page = 0;
} }
ndbout << endl; ndbout << endl;
}//for }//for
...@@ -310,93 +325,6 @@ Uint32 readFromFile(FILE * f, Uint32 *toPtr, Uint32 sizeInWords) { ...@@ -310,93 +325,6 @@ Uint32 readFromFile(FILE * f, Uint32 *toPtr, Uint32 sizeInWords) {
} }
//----------------------------------------------------------------
//
//----------------------------------------------------------------
Uint32 readRecordOverPageBoundary(Uint32 *pagePtr, Uint32 pageIndex, Uint32 oldPageIndex, Uint32 recordType) {
Uint32 pageHeader[PAGEHEADERSIZE];
Uint32 tmpPages[PAGESIZE*10];
PageHeader *thePageHeader;
Uint32 recordSize = 0;
PrepareOperationRecord *poRecord;
CommitTransactionRecord *ctRecord;
InvalidCommitTransactionRecord *ictRecord;
memcpy(pageHeader, pagePtr + PAGESIZE, PAGEHEADERSIZE*sizeof(Uint32));
memcpy(tmpPages, pagePtr + oldPageIndex, (PAGESIZE - oldPageIndex)*sizeof(Uint32));
memcpy(tmpPages + PAGESIZE - oldPageIndex ,
(pagePtr + PAGESIZE + PAGEHEADERSIZE),
(PAGESIZE - PAGEHEADERSIZE)*sizeof(Uint32));
switch(recordType) {
case ZPREP_OP_TYPE:
poRecord = (PrepareOperationRecord *) tmpPages;
recordSize = poRecord->getLogRecordSize();
if (recordSize < (PAGESIZE - PAGEHEADERSIZE)) {
if (theCheckFlag) {
if(!poRecord->check()) {
ndbout << "Error in poRecord->check() (readRecordOverPageBoundary)" << endl;
doExit();
}
}
if (thePrintFlag) ndbout << (*poRecord);
} else {
ndbout << "Error: Record greater than a Page" << endl;
}
break;
case ZCOMMIT_TYPE:
ctRecord = (CommitTransactionRecord *) tmpPages;
recordSize = ctRecord->getLogRecordSize();
if (recordSize < (PAGESIZE - PAGEHEADERSIZE)) {
if (theCheckFlag) {
if(!ctRecord->check()) {
ndbout << "Error in ctRecord->check() (readRecordOverPageBoundary)" << endl;
doExit();
}
}
if (thePrintFlag) ndbout << (*ctRecord);
} else {
ndbout << endl << "Error: Record greater than a Page" << endl;
}
break;
case ZINVALID_COMMIT_TYPE:
ictRecord = (InvalidCommitTransactionRecord *) tmpPages;
recordSize = ictRecord->getLogRecordSize();
if (recordSize < (PAGESIZE - PAGEHEADERSIZE)) {
if (theCheckFlag) {
if(!ictRecord->check()) {
ndbout << "Error in ictRecord->check() (readRecordOverPageBoundary)" << endl;
doExit();
}
}
if (thePrintFlag) ndbout << (*ictRecord);
} else {
ndbout << endl << "Error: Record greater than a Page" << endl;
}
break;
case ZNEW_PREP_OP_TYPE:
case ZABORT_TYPE:
case ZFRAG_SPLIT_TYPE:
case ZNEXT_MBYTE_TYPE:
ndbout << endl << "Record type = " << recordType << " not implemented." << endl;
return 0;
default:
ndbout << endl << "Error: Unknown record type. Record type = " << recordType << endl;
return 0;
}
thePageHeader = (PageHeader *) (pagePtr + PAGESIZE);
if (thePrintFlag) ndbout << (*thePageHeader);
return PAGEHEADERSIZE - PAGESIZE + oldPageIndex + recordSize;
}
//---------------------------------------------------------------- //----------------------------------------------------------------
// //
//---------------------------------------------------------------- //----------------------------------------------------------------
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7