Link the current page to newPageID:newOffset and then write it to the pager. The link destination could be a new page at the end of the queue, or the beginning of an existing chain of queue pages. If initializeNewPage is true a page buffer will be allocated for the new page and it will be initialized as a new tail page. if initializeExtentInfo is true in addition to initializeNewPage, update the e
| 757 | // if initializeExtentInfo is true in addition to initializeNewPage, update the extentEndPageID info |
| 758 | // in the mew page being added using newExtentPage and prevExtentEndPageID parameters |
| 759 | void addNewPage(PhysicalPageID newPageID, |
| 760 | int newOffset, |
| 761 | bool initializeNewPage, |
| 762 | bool initializeExtentInfo = false, |
| 763 | bool newExtentPage = false, |
| 764 | PhysicalPageID prevExtentEndPageID = invalidPhysicalPageID) { |
| 765 | ASSERT(mode == WRITE); |
| 766 | ASSERT(newPageID != invalidPhysicalPageID); |
| 767 | debug_printf("FIFOQueue::Cursor(%s) Adding page %s initPage=%d initExtentInfo=%d newExtentPage=%d\n", |
| 768 | toString().c_str(), |
| 769 | ::toString(newPageID).c_str(), |
| 770 | initializeNewPage, |
| 771 | initializeExtentInfo, |
| 772 | newExtentPage); |
| 773 | |
| 774 | // Update existing page/newLastPageID and write, if it exists |
| 775 | if (page) { |
| 776 | setNext(newPageID, newOffset); |
| 777 | debug_printf("FIFOQueue::Cursor(%s) Linked new page %s:%d\n", |
| 778 | toString().c_str(), |
| 779 | ::toString(newPageID).c_str(), |
| 780 | newOffset); |
| 781 | writePage(); |
| 782 | prevExtentEndPageID = header()->extentEndPageID; |
| 783 | if (pageID == prevExtentEndPageID) |
| 784 | newExtentPage = true; |
| 785 | debug_printf( |
| 786 | "FIFOQueue::Cursor(%s) Linked new page. pageID %u, newPageID %u, prevExtentEndPageID %u\n", |
| 787 | toString().c_str(), |
| 788 | pageID, |
| 789 | newPageID, |
| 790 | prevExtentEndPageID); |
| 791 | } |
| 792 | |
| 793 | pageID = newPageID; |
| 794 | offset = newOffset; |
| 795 | |
| 796 | if (initializeNewPage) { |
| 797 | debug_printf("FIFOQueue::Cursor(%s) Initializing new page. usesExtents: %d, initializeExtentInfo: %d\n", |
| 798 | toString().c_str(), |
| 799 | queue->usesExtents, |
| 800 | initializeExtentInfo); |
| 801 | page = queue->pager->newPageBuffer(); |
| 802 | page->init(EncodingType::XXHash64, |
| 803 | queue->usesExtents ? PageType::QueuePageInExtent : PageType::QueuePageStandalone, |
| 804 | (uint8_t)queue->queueID); |
| 805 | setNext(0, 0); |
| 806 | auto p = header(); |
| 807 | ASSERT(newOffset == 0); |
| 808 | p->endOffset = 0; |
| 809 | p->itemSpace = page->dataSize() - sizeof(QueuePage); |
| 810 | if (g_network->isSimulated() && deterministicRandom()->coinflip()) { |
| 811 | // Randomly reduce available item space to cause more queue pages to be needed |
| 812 | int reducedSpace = deterministicRandom()->randomInt(50, p->itemSpace); |
| 813 | |
| 814 | // Zero the eliminated space |
| 815 | memset(header()->begin() + reducedSpace, 0, p->itemSpace - reducedSpace); |
| 816 |