Method readNext

fdbserver/VersionedBTree.actor.cpp:1013–1116

Read the next item from the cursor, possibly moving to and waiting for a new page if the prior page was exhausted. If the item is <= inclusiveMaximum, then return it after advancing the cursor to the next item. Otherwise, return nothing and do not advance the cursor. If locked is true, this call owns the mutex, which would have been locked by readNext() before a recursive call. See waitThenReadNex

		// If locked is true, this call owns the mutex, which would have been locked by readNext() before a recursive
		// call. See waitThenReadNext() for more detail.
		Future<Optional<T>> readNext(const Optional<T>& inclusiveMaximum = {}, FlowMutex::Lock* lock = nullptr) {
			if ((mode != POP && mode != READONLY) || pageID == invalidPhysicalPageID || pageID == endPageID) {
				debug_printf("FIFOQueue::Cursor(%s) readNext returning nothing\n", toString().c_str());
				return Optional<T>();
			}

			// If we don't have a lock and the mutex isn't available then acquire it
			if (lock == nullptr && isBusy()) {
				return waitThenReadNext(this, inclusiveMaximum, lock, false);
			}

			// We now know pageID is valid and should be used, but page might not point to it yet
			if (!page) {
				debug_printf("FIFOQueue::Cursor(%s) loading\n", toString().c_str());

				// If the next pageID loading or loaded is not the page we should be reading then restart the load
				// nextPageID coud be different because it could be invalid or it could be no longer relevant
				// if the previous commit added new pages to the front of the queue.
				if (pageID != nextPageID) {
					debug_printf("FIFOQueue::Cursor(%s) reloading\n", toString().c_str());
					startNextPageLoad(pageID);
				}

				if (!nextPageReader.isReady()) {
					return waitThenReadNext(this, inclusiveMaximum, lock, true);
				}

				page = nextPageReader.get();

				// Start loading the next page if it's not the end page
				auto p = header();
				if (p->nextPageID != endPageID) {
					startNextPageLoad(p->nextPageID);
				} else {
					// Prevent a future next page read from reusing the same result as page would have to be updated
					// before the queue would read it again
					nextPageID = invalidPhysicalPageID;
				}
			}
			auto p = header();
			debug_printf("FIFOQueue::Cursor(%s) readNext reading at current position\n", toString().c_str());
			ASSERT(offset < p->endOffset);
			int bytesRead;
			const T result = Codec::readFromBytes(p->begin() + offset, bytesRead);

			if (inclusiveMaximum.present() && inclusiveMaximum.get() < result) {
				debug_printf("FIFOQueue::Cursor(%s) not popping %s, exceeds upper bound %s\n",
				             toString().c_str(),
				             ::toString(result).c_str(),
				             ::toString(inclusiveMaximum.get()).c_str());

				return Optional<T>();
			}

			offset += bytesRead;
			if (mode == POP) {
				--queue->numEntries;
			}