Method Manager

src/jrd/replication/Manager.cpp:99–184

// Replication manager

Manager::Manager(const string& dbId,
				 const Replication::Config* config)
	: m_config(config),
	  m_replicas(getPool()),
	  m_buffers(getPool()),
	  m_queue(getPool()),
	  m_queueSize(0),
	  m_sequence(0),
	  m_shutdown(false),
	  m_signalled(false)
{
	// Startup the journalling

	const auto tdbb = JRD_get_thread_data();
	const auto dbb = tdbb->getDatabase();

	dbb->ensureGuid(tdbb);
	const Guid& guid = dbb->dbb_guid;

	if (config->journalDirectory.hasData())
	{
		// At this point it is unknown if change log shared memory exists or not.
		// To avoid race condition with concurrent changing of current replication
		// sequence, take and hold shared lock on header page while creating
		// ChangeLog instance.

		WIN window(HEADER_PAGE_NUMBER);
		CCH_FETCH(tdbb, &window, LCK_read, pag_header);

		Cleanup releaseHeader([&] {
			CCH_RELEASE(tdbb, &window);
		});

		// Call below will fetch header page with LCK_read lock, it is allowed and OK.
		m_sequence = dbb->getReplSequence(tdbb);

		m_changeLog = FB_NEW_POOL(getPool())
			ChangeLog(getPool(), dbId, guid, m_sequence, config);
	}
	else
		fb_assert(config->syncReplicas.hasData());

	// Attach to synchronous replicas (if any)

	FbLocalStatus localStatus;
	DispatcherPtr provider;

	for (const auto& iter : m_config->syncReplicas)
	{
		ClumpletWriter dpb(ClumpletReader::dpbList, MAX_DPB_SIZE);
		dpb.insertByte(isc_dpb_no_db_triggers, 1);

		if (iter.username.hasData())
		{
			dpb.insertString(isc_dpb_user_name, iter.username);

			if (iter.password.hasData())
				dpb.insertString(isc_dpb_password, iter.password);