postgresql/src/backend/replication/walreceiverfuncs.c

/*-------------------------------------------------------------------------
 *
 * walreceiverfuncs.c
 *
 * This file contains functions used by the startup process to communicate
 * with the walreceiver process. Functions implementing walreceiver itself
 * are in walreceiver.c.
 *
 * Portions Copyright (c) 2010-2010, PostgreSQL Global Development Group
 *
 *
 * IDENTIFICATION
 *	  $PostgreSQL: pgsql/src/backend/replication/walreceiverfuncs.c,v 1.2 2010/01/20 09:16:24 heikki Exp $
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <signal.h>

#include "access/xlog_internal.h"
#include "replication/walreceiver.h"
#include "storage/fd.h"
#include "storage/pmsignal.h"
#include "storage/shmem.h"
#include "utils/guc.h"

WalRcvData *WalRcv = NULL;

static bool CheckForStandbyTrigger(void);
static void ShutdownWalRcv(void);

/* Report shared memory space needed by WalRcvShmemInit */
Size
WalRcvShmemSize(void)
{
	Size size = 0;

	size = add_size(size, sizeof(WalRcvData));

	return size;
}

/* Allocate and initialize walreceiver-related shared memory */
void
WalRcvShmemInit(void)
{
	bool	found;

	WalRcv = (WalRcvData *)
		ShmemInitStruct("Wal Receiver Ctl", WalRcvShmemSize(), &found);

	if (WalRcv == NULL)
		ereport(FATAL,
				(errcode(ERRCODE_OUT_OF_MEMORY),
				 errmsg("not enough shared memory for walreceiver")));
	if (found)
		return;					/* already initialized */

	/* Initialize the data structures */
	MemSet(WalRcv, 0, WalRcvShmemSize());
	WalRcv->walRcvState = WALRCV_NOT_STARTED;
	SpinLockInit(&WalRcv->mutex);
}

/* Is walreceiver in progress (or starting up)? */
bool
WalRcvInProgress(void)
{
	/* use volatile pointer to prevent code rearrangement */
	volatile WalRcvData *walrcv = WalRcv;
	WalRcvState state;

	SpinLockAcquire(&walrcv->mutex);
	state = walrcv->walRcvState;
	SpinLockRelease(&walrcv->mutex);

	if (state == WALRCV_RUNNING || state == WALRCV_STOPPING)
		return true;
	else
		return false;
}

/*
 * Wait for the XLOG record at given position to become available.
 *
 * 'recptr' indicates the byte position which caller wants to read the
 * XLOG record up to. The byte position actually written and flushed
 * by walreceiver is returned. It can be higher than the requested
 * location, and the caller can safely read up to that point without
 * calling WaitNextXLogAvailable() again.
 *
 * If WAL streaming is ended (because a trigger file is found), *finished
 * is set to true and function returns immediately. The returned position
 * can be lower than requested in that case.
 *
 * Called by the startup process during streaming recovery.
 */
XLogRecPtr
WaitNextXLogAvailable(XLogRecPtr recptr, bool *finished)
{
	static XLogRecPtr receivedUpto = {0, 0};

	*finished = false;

	/* Quick exit if already known available */
	if (XLByteLT(recptr, receivedUpto))
		return receivedUpto;

	for (;;)
	{
		/* use volatile pointer to prevent code rearrangement */
		volatile WalRcvData *walrcv = WalRcv;

		/* Update local status */
		SpinLockAcquire(&walrcv->mutex);
		receivedUpto = walrcv->receivedUpto;
		SpinLockRelease(&walrcv->mutex);

		/* If available already, leave here */
		if (XLByteLT(recptr, receivedUpto))
			return receivedUpto;

		/* Check to see if the trigger file exists */
		if (CheckForStandbyTrigger())
		{
			*finished = true;
			return receivedUpto;
		}

		pg_usleep(100000L); /* 100ms */

		/*
		 * This possibly-long loop needs to handle interrupts of startup
		 * process.
		 */
		HandleStartupProcInterrupts();

		/*
		 * Emergency bailout if postmaster has died.  This is to avoid the
		 * necessity for manual cleanup of all postmaster children.
		 */
		if (!PostmasterIsAlive(true))
			exit(1);
	}
}

/*
 * Stop walreceiver and wait for it to die.
 */
static void
ShutdownWalRcv(void)
{
	/* use volatile pointer to prevent code rearrangement */
	volatile WalRcvData *walrcv = WalRcv;
	pid_t walrcvpid;

	/*
	 * Request walreceiver to stop. Walreceiver will switch to WALRCV_STOPPED
	 * mode once it's finished, and will also request postmaster to not
	 * restart itself.
	 */
	SpinLockAcquire(&walrcv->mutex);
	Assert(walrcv->walRcvState == WALRCV_RUNNING);
	walrcv->walRcvState = WALRCV_STOPPING;
	walrcvpid = walrcv->pid;
	SpinLockRelease(&walrcv->mutex);

	/*
	 * Pid can be 0, if no walreceiver process is active right now.
	 * Postmaster should restart it, and when it does, it will see the
	 * STOPPING state.
	 */
	if (walrcvpid != 0)
		kill(walrcvpid, SIGTERM);

	/*
	 * Wait for walreceiver to acknowledge its death by setting state to
	 * WALRCV_STOPPED.
	 */
	while (WalRcvInProgress())
	{
		/*
		 * This possibly-long loop needs to handle interrupts of startup
		 * process.
		 */
		HandleStartupProcInterrupts();

		pg_usleep(100000);		/* 100ms */
	}
}

/*
 * Check to see if the trigger file exists. If it does, request postmaster
 * to shut down walreceiver and wait for it to exit, and remove the trigger
 * file.
 */
static bool
CheckForStandbyTrigger(void)
{
	struct stat stat_buf;

	if (TriggerFile == NULL)
		return false;

	if (stat(TriggerFile, &stat_buf) == 0)
	{
		ereport(LOG,
				(errmsg("trigger file found: %s", TriggerFile)));
		ShutdownWalRcv();
		unlink(TriggerFile);
		return true;
	}
	return false;
}

/*
 * Request postmaster to start walreceiver.
 *
 * recptr indicates the position where streaming should begin, and conninfo
 * is a libpq connection string to use.
 */
void
RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo)
{
	/* use volatile pointer to prevent code rearrangement */
	volatile WalRcvData *walrcv = WalRcv;

	Assert(walrcv->walRcvState == WALRCV_NOT_STARTED);

	/* locking is just pro forma here; walreceiver isn't started yet */
	SpinLockAcquire(&walrcv->mutex);
	walrcv->receivedUpto = recptr;
	if (conninfo != NULL)
		strlcpy((char *) walrcv->conninfo, conninfo, MAXCONNINFO);
	else
		walrcv->conninfo[0] = '\0';
	walrcv->walRcvState = WALRCV_RUNNING;
	SpinLockRelease(&walrcv->mutex);

	SendPostmasterSignal(PMSIGNAL_START_WALRECEIVER);
}

/*
 * Returns the byte position that walreceiver has written
 */
XLogRecPtr
GetWalRcvWriteRecPtr(void)
{
	/* use volatile pointer to prevent code rearrangement */
	volatile WalRcvData *walrcv = WalRcv;
	XLogRecPtr	recptr;

	SpinLockAcquire(&walrcv->mutex);
	recptr = walrcv->receivedUpto;
	SpinLockRelease(&walrcv->mutex);

	return recptr;
}