Method write_impl

fdbserver/VersionedBTree.actor.cpp:856–952

		// Write item to the next position in the current page or, if it won't fit, add a new page and write it there.
		ACTOR static Future<Void> write_impl(Cursor* self, T item) {
			ASSERT(self->mode == WRITE);

			state FlowMutex::Lock lock;
			state bool mustWait = self->isBusy();
			state int bytesNeeded = Codec::bytesNeeded(item);
			state bool needNewPage =
			    self->pageID == invalidPhysicalPageID || self->offset + bytesNeeded > self->header()->itemSpace;

			if (g_network->isSimulated()) {
				// Sometimes (1% probability) decide a new page is needed as long as at least 1 item has been
				// written (indicated by non-zero offset) to the current page.
				if ((self->offset > 0) && deterministicRandom()->random01() < 0.01) {
					needNewPage = true;
				}
			}

			debug_printf("FIFOQueue::Cursor(%s) write(%s) mustWait=%d needNewPage=%d\n",
			             self->toString().c_str(),
			             ::toString(item).c_str(),
			             mustWait,
			             needNewPage);

			// If we have to wait for the mutex because it's busy, or we need a new page, then wait for the mutex.
			if (mustWait || needNewPage) {
				FlowMutex::Lock _lock = wait(self->mutex.take());
				lock = _lock;

				// If we had to wait because the mutex was busy, then update needNewPage as another writer
				// would have changed the cursor state
				// Otherwise, taking the mutex would be immediate so no other writer could have run
				if (mustWait) {
					needNewPage =
					    self->pageID == invalidPhysicalPageID || self->offset + bytesNeeded > self->header()->itemSpace;
					if (g_network->isSimulated()) {
						// Sometimes (1% probability) decide a new page is needed as long as at least 1 item has been
						// written (indicated by non-zero offset) to the current page.
						if ((self->offset > 0) && deterministicRandom()->random01() < 0.01) {
							needNewPage = true;
						}
					}
				}
			}

			// If we need a new page, add one.
			if (needNewPage) {
				debug_printf("FIFOQueue::Cursor(%s) write(%s) page is full, adding new page\n",
				             self->toString().c_str(),
				             ::toString(item).c_str());
				state PhysicalPageID newPageID;
				// If this is an extent based queue, check if there is an available page in current extent
				if (self->queue->usesExtents) {
					bool allocateNewExtent = false;
					if (self->pageID != invalidPhysicalPageID) {
						auto praw = self->header();
						if (praw->extentCurPageID < praw->extentEndPageID) {
							newPageID = praw->extentCurPageID + 1;
						} else {