Method flush

src/jrd/replication/Manager.cpp:253–339

}

void Manager::flush(UCharBuffer* buffer, bool sync, bool prepare)
{
	fb_assert(!m_shutdown);
	fb_assert(buffer && buffer->hasData());

	const auto prepareBuffer = prepare ? buffer : nullptr;

	MutexLockGuard guard(m_queueMutex, FB_FUNCTION);

	// Add the current chunk to the queue
	m_queue.add(buffer);
	m_queueSize += buffer->getCount();

	// If the background thread is lagging too far behind,
	// replicate packets synchronously rather than relying
	// on the background thread to catch up any time soon
	const bool lagging = (m_queueSize > MAX_BG_WRITER_LAG);

	if (sync || prepare || lagging)
	{
		const auto tdbb = JRD_get_thread_data();
		const auto dbb = tdbb->getDatabase();

		for (auto& buffer : m_queue)
		{
			if (buffer)
			{
				auto length = (ULONG) buffer->getCount();
				fb_assert(length);
				bool hasData = true;

				if (m_changeLog)
				{
					if (prepareBuffer == buffer)
					{
						// Remove the opPrepareTransaction command from the journal
						fb_assert(buffer->back() == opPrepareTransaction);

						const auto block = (Block*) buffer->begin();
						block->length -= sizeof(UCHAR);
						length -= sizeof(UCHAR);
						hasData = (block->length != 0);
					}

					if (hasData)
					{
						const auto sequence = m_changeLog->write(length, buffer->begin(), sync);

						if (sequence != m_sequence)
						{
							dbb->setReplSequence(tdbb, sequence);
							m_sequence = sequence;
						}
					}

					if (prepareBuffer == buffer)
					{
						const auto block = (Block*) buffer->begin();