postgresql/src/backend/utils/cache/inval.c

/*-------------------------------------------------------------------------
 *
 * inval.c
 *	  POSTGRES cache invalidation dispatcher code.
 *
 *	This is subtle stuff, so pay attention:
 *
 *	When a tuple is updated or deleted, our time qualification rules consider
 *	that it is *still valid* so long as we are in the same command, ie,
 *	until the next CommandCounterIncrement() or transaction commit.
 *	(See utils/time/tqual.c.)  At the command boundary, the old tuple stops
 *	being valid and the new version, if any, becomes valid.  Therefore,
 *	we cannot simply flush a tuple from the system caches during heap_update()
 *	or heap_delete().  The tuple is still good at that point; what's more,
 *	even if we did flush it, it might be reloaded into the caches by a later
 *	request in the same command.  So the correct behavior is to keep a list
 *	of outdated (updated/deleted) tuples and then do the required cache
 *	flushes at the next command boundary.  Similarly, we need a list of
 *	inserted tuples (including new versions of updated tuples), which we will
 *	use to flush those tuples out of the caches if we abort the transaction.
 *	Notice that the first list lives only till command boundary, whereas the
 *	second lives till end of transaction.  Finally, we need a third list of
 *	all tuples outdated in the current transaction; if we commit, we send
 *	those invalidation events to all other backends (via the SI message queue)
 *	so that they can flush obsolete entries from their caches.	This list
 *	definitely can't be processed until after we commit, otherwise the other
 *	backends won't see our updated tuples as good.
 *
 *	We do not need to register EVERY tuple operation in this way, just those
 *	on tuples in relations that have associated catcaches.	We do, however,
 *	have to register every operation on every tuple that *could* be in a
 *	catcache, whether or not it currently is in our cache.	Also, if the
 *	tuple is in a relation that has multiple catcaches, we need to register
 *	an invalidation message for each such catcache.  catcache.c's
 *	PrepareToInvalidateCacheTuple() routine provides the knowledge of which
 *	catcaches may need invalidation for a given tuple.
 *
 *	Also, whenever we see an operation on a pg_class or pg_attribute tuple,
 *	we register a relcache flush operation for the relation described by that
 *	tuple.
 *
 *	We keep the relcache flush requests in lists separate from the catcache
 *	tuple flush requests.  This allows us to issue all the pending catcache
 *	flushes before we issue relcache flushes, which saves us from loading
 *	a catcache tuple during relcache load only to flush it again right away.
 *	Also, we avoid queuing multiple relcache flush requests for the same
 *	relation, since a relcache flush is relatively expensive to do.
 *	(XXX is it worth testing likewise for duplicate catcache flush entries?
 *	Probably not.)
 *
 *	All the request lists are kept in TopTransactionContext memory, since
 *	they need not live beyond the end of the current transaction.
 *
 *
 * Portions Copyright (c) 1996-2001, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  $Header: /cvsroot/pgsql/src/backend/utils/cache/inval.c,v 1.47 2001/11/16 23:30:35 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "catalog/catalog.h"
#include "miscadmin.h"
#include "storage/sinval.h"
#include "utils/catcache.h"
#include "utils/inval.h"
#include "utils/memutils.h"
#include "utils/relcache.h"


/*
 * To minimize palloc traffic, we keep pending requests in successively-
 * larger chunks (a slightly more sophisticated version of an expansible
 * array).	All request types can be stored as SharedInvalidationMessage
 * records.
 */
typedef struct InvalidationChunk
{
	struct InvalidationChunk *next;		/* list link */
	int			nitems;			/* # items currently stored in chunk */
	int			maxitems;		/* size of allocated array in this chunk */
	SharedInvalidationMessage msgs[1];	/* VARIABLE LENGTH ARRAY */
} InvalidationChunk;			/* VARIABLE LENGTH STRUCTURE */

typedef struct InvalidationListHeader
{
	InvalidationChunk *cclist;	/* list of chunks holding catcache msgs */
	InvalidationChunk *rclist;	/* list of chunks holding relcache msgs */
} InvalidationListHeader;

/*
 * ----------------
 *	Invalidation info is divided into three parts.
 *	1) shared invalidation to be sent to all backends at commit
 *	2) local invalidation for the transaction itself (actually, just
 *	   for the current command within the transaction)
 *	3) rollback information for the transaction itself (in case we abort)
 * ----------------
 */

/*
 * head of invalidation message list for all backends
 * eaten by AtCommit_Cache() in CommitTransaction()
 */
static InvalidationListHeader GlobalInvalidMsgs;

/*
 * head of invalidation message list for the current command
 * eaten by AtCommit_LocalCache() in CommandCounterIncrement()
 */
static InvalidationListHeader LocalInvalidMsgs;

/*
 * head of rollback message list for abort-time processing
 * eaten by AtAbort_Cache() in AbortTransaction()
 */
static InvalidationListHeader RollbackMsgs;


/* ----------------------------------------------------------------
 *				Invalidation list support functions
 *
 * These three routines encapsulate processing of the "chunked"
 * representation of what is logically just a list of messages.
 * ----------------------------------------------------------------
 */

/*
 * AddInvalidationMessage
 *		Add an invalidation message to a list (of chunks).
 *
 * Note that we do not pay any great attention to maintaining the original
 * ordering of the messages.
 */
static void
AddInvalidationMessage(InvalidationChunk **listHdr,
					   SharedInvalidationMessage *msg)
{
	InvalidationChunk *chunk = *listHdr;

	if (chunk == NULL)
	{
		/* First time through; create initial chunk */
#define FIRSTCHUNKSIZE 16
		chunk = (InvalidationChunk *)
			MemoryContextAlloc(TopTransactionContext,
							   sizeof(InvalidationChunk) +
				(FIRSTCHUNKSIZE - 1) *sizeof(SharedInvalidationMessage));
		chunk->nitems = 0;
		chunk->maxitems = FIRSTCHUNKSIZE;
		chunk->next = *listHdr;
		*listHdr = chunk;
	}
	else if (chunk->nitems >= chunk->maxitems)
	{
		/* Need another chunk; double size of last chunk */
		int			chunksize = 2 * chunk->maxitems;

		chunk = (InvalidationChunk *)
			MemoryContextAlloc(TopTransactionContext,
							   sizeof(InvalidationChunk) +
					 (chunksize - 1) *sizeof(SharedInvalidationMessage));
		chunk->nitems = 0;
		chunk->maxitems = chunksize;
		chunk->next = *listHdr;
		*listHdr = chunk;
	}
	/* Okay, add message to current chunk */
	chunk->msgs[chunk->nitems] = *msg;
	chunk->nitems++;
}

/*
 * Free a list of inval message chunks.
 *
 * NOTE: when we are about to commit or abort a transaction, it's
 * not really necessary to pfree the lists explicitly, since they will
 * go away anyway when TopTransactionContext is destroyed.
 */
static void
FreeInvalidationMessageList(InvalidationChunk **listHdr)
{
	InvalidationChunk *chunk = *listHdr;

	*listHdr = NULL;

	while (chunk != NULL)
	{
		InvalidationChunk *nextchunk = chunk->next;

		pfree(chunk);
		chunk = nextchunk;
	}
}

/*
 * Process a list of invalidation messages.
 *
 * This is a macro that executes the given code fragment for each message in
 * a message chunk list.  The fragment should refer to the message as *msg.
 */
#define ProcessMessageList(listHdr, codeFragment) \
	do { \
		InvalidationChunk *_chunk; \
		for (_chunk = (listHdr); _chunk != NULL; _chunk = _chunk->next) \
		{ \
			int		_cindex; \
			for (_cindex = 0; _cindex < _chunk->nitems; _cindex++) \
			{ \
				SharedInvalidationMessage *msg = &_chunk->msgs[_cindex]; \
				codeFragment; \
			} \
		} \
	} while (0)


/* ----------------------------------------------------------------
 *				Invalidation set support functions
 *
 * These routines understand about the division of a logical invalidation
 * list into separate physical lists for catcache and relcache entries.
 * ----------------------------------------------------------------
 */

/*
 * Add a catcache inval entry
 */
static void
AddCatcacheInvalidationMessage(InvalidationListHeader *hdr,
							   int id, Index hashIndex,
							   ItemPointer tuplePtr, Oid dbId)
{
	SharedInvalidationMessage msg;

	msg.cc.id = (int16) id;
	msg.cc.hashIndex = (uint16) hashIndex;
	msg.cc.dbId = dbId;
	msg.cc.tuplePtr = *tuplePtr;
	AddInvalidationMessage(&hdr->cclist, &msg);
}

/*
 * Add a relcache inval entry
 */
static void
AddRelcacheInvalidationMessage(InvalidationListHeader *hdr,
							   Oid dbId, Oid relId)
{
	SharedInvalidationMessage msg;

	/* Don't add a duplicate item */
	/* We assume comparing relId is sufficient, needn't check dbId */
	ProcessMessageList(hdr->rclist,
					   if (msg->rc.relId == relId) return);

	/* OK, add the item */
	msg.rc.id = SHAREDINVALRELCACHE_ID;
	msg.rc.dbId = dbId;
	msg.rc.relId = relId;
	AddInvalidationMessage(&hdr->rclist, &msg);
}

/*
 * Reset an invalidation list to empty
 *
 * physicalFree may be set false if caller knows transaction is ending
 */
static void
DiscardInvalidationMessages(InvalidationListHeader *hdr, bool physicalFree)
{
	if (physicalFree)
	{
		/* Physically pfree the list data */
		FreeInvalidationMessageList(&hdr->cclist);
		FreeInvalidationMessageList(&hdr->rclist);
	}
	else
	{
		/*
		 * Assume the storage will go away at xact end, just reset
		 * pointers
		 */
		hdr->cclist = NULL;
		hdr->rclist = NULL;
	}
}

/*
 * Execute the given function for all the messages in an invalidation list.
 * The list is not altered.
 *
 * catcache entries are processed first, for reasons mentioned above.
 */
static void
ProcessInvalidationMessages(InvalidationListHeader *hdr,
							void (*func) (SharedInvalidationMessage *msg))
{
	ProcessMessageList(hdr->cclist, func(msg));
	ProcessMessageList(hdr->rclist, func(msg));
}

/* ----------------------------------------------------------------
 *					  private support functions
 * ----------------------------------------------------------------
 */

/*
 * RegisterCatcacheInvalidation
 *
 * Register an invalidation event for an updated/deleted catcache entry.
 * We insert the event into both GlobalInvalidMsgs (for transmission
 * to other backends at transaction commit) and LocalInvalidMsgs (for
 * my local invalidation at end of command within xact).
 */
static void
RegisterCatcacheInvalidation(int cacheId,
							 Index hashIndex,
							 ItemPointer tuplePtr,
							 Oid dbId)
{
	AddCatcacheInvalidationMessage(&GlobalInvalidMsgs,
								   cacheId, hashIndex, tuplePtr, dbId);
	AddCatcacheInvalidationMessage(&LocalInvalidMsgs,
								   cacheId, hashIndex, tuplePtr, dbId);
}

/*
 * RegisterRelcacheInvalidation
 *
 * As above, but register a relcache invalidation event.
 */
static void
RegisterRelcacheInvalidation(Oid dbId, Oid relId)
{
	AddRelcacheInvalidationMessage(&GlobalInvalidMsgs,
								   dbId, relId);
	AddRelcacheInvalidationMessage(&LocalInvalidMsgs,
								   dbId, relId);
}

/*
 * RegisterCatcacheRollback
 *
 * Register an invalidation event for an inserted catcache entry.
 * This only needs to be flushed out of my local catcache, if I abort.
 */
static void
RegisterCatcacheRollback(int cacheId,
						 Index hashIndex,
						 ItemPointer tuplePtr,
						 Oid dbId)
{
	AddCatcacheInvalidationMessage(&RollbackMsgs,
								   cacheId, hashIndex, tuplePtr, dbId);
}

/*
 * RegisterRelcacheRollback
 *
 * As above, but register a relcache invalidation event.
 */
static void
RegisterRelcacheRollback(Oid dbId, Oid relId)
{
	AddRelcacheInvalidationMessage(&RollbackMsgs,
								   dbId, relId);
}

/*
 * LocalExecuteInvalidationMessage
 *
 * Process a single invalidation message (which could be either type).
 * Only the local caches are flushed; this does not transmit the message
 * to other backends.
 */
static void
LocalExecuteInvalidationMessage(SharedInvalidationMessage *msg)
{
	if (msg->id >= 0)
	{
		if (msg->cc.dbId == MyDatabaseId || msg->cc.dbId == 0)
			CatalogCacheIdInvalidate(msg->cc.id,
									 msg->cc.hashIndex,
									 &msg->cc.tuplePtr);
	}
	else if (msg->id == SHAREDINVALRELCACHE_ID)
	{
		if (msg->rc.dbId == MyDatabaseId || msg->rc.dbId == 0)
			RelationIdInvalidateRelationCacheByRelationId(msg->rc.relId);
	}
	else
	{
		elog(FATAL, "ExecuteInvalidationMessage: bogus message id %d",
			 msg->id);
	}
}

/*
 *		InvalidateSystemCaches
 *
 *		This blows away all tuples in the system catalog caches and
 *		all the cached relation descriptors (and closes their files too).
 *		Relation descriptors that have positive refcounts are then rebuilt.
 *
 *		We call this when we see a shared-inval-queue overflow signal,
 *		since that tells us we've lost some shared-inval messages and hence
 *		don't know what needs to be invalidated.
 */
static void
InvalidateSystemCaches(void)
{
	ResetCatalogCaches();
	RelationCacheInvalidate();
}

/*
 * PrepareForTupleInvalidation
 *		Invoke functions for the tuple which register invalidation
 *		of catalog/relation cache.
 */
static void
PrepareForTupleInvalidation(Relation relation, HeapTuple tuple,
							void (*CacheIdRegisterFunc) (int, Index,
													   ItemPointer, Oid),
							void (*RelationIdRegisterFunc) (Oid, Oid))
{
	Oid			tupleRelId;
	Oid			relationId;

	if (IsBootstrapProcessingMode())
		return;

	/*
	 * We only need to worry about invalidation for tuples that are in
	 * system relations; user-relation tuples are never in catcaches and
	 * can't affect the relcache either.
	 *
	 * TOAST tuples can likewise be ignored here.
	 */
	if (!IsSystemRelationName(NameStr(RelationGetForm(relation)->relname)))
		return;
	if (IsToastRelationName(NameStr(RelationGetForm(relation)->relname)))
		return;

	/*
	 * First let the catcache do its thing
	 */
	PrepareToInvalidateCacheTuple(relation, tuple,
								  CacheIdRegisterFunc);

	/*
	 * Now, is this tuple one of the primary definers of a relcache entry?
	 */
	tupleRelId = RelationGetRelid(relation);

	if (tupleRelId == RelOid_pg_class)
		relationId = tuple->t_data->t_oid;
	else if (tupleRelId == RelOid_pg_attribute)
		relationId = ((Form_pg_attribute) GETSTRUCT(tuple))->attrelid;
	else
		return;

	/*
	 * Yes.  We need to register a relcache invalidation event for the
	 * relation identified by relationId.
	 *
	 * KLUGE ALERT: we always send the relcache event with MyDatabaseId, even
	 * if the rel in question is shared.  This essentially means that only
	 * backends in this same database will react to the relcache flush
	 * request.  This is in fact appropriate, since only those backends
	 * could see our pg_class or pg_attribute change anyway.  It looks a
	 * bit ugly though.
	 */
	(*RelationIdRegisterFunc) (MyDatabaseId, relationId);
}


/* ----------------------------------------------------------------
 *					  public functions
 * ----------------------------------------------------------------
 */

/*
 * AcceptInvalidationMessages
 *		Read and process invalidation messages from the shared invalidation
 *		message queue.
 *
 * Note:
 *		This should be called as the first step in processing a transaction.
 */
void
AcceptInvalidationMessages(void)
{
	ReceiveSharedInvalidMessages(LocalExecuteInvalidationMessage,
								 InvalidateSystemCaches);
}

/*
 * AtEOXactInvalidationMessages
 *		Process queued-up invalidation messages at end of transaction.
 *
 * If isCommit, we must send out the messages in our GlobalInvalidMsgs list
 * to the shared invalidation message queue.  Note that these will be read
 * not only by other backends, but also by our own backend at the next
 * transaction start (via AcceptInvalidationMessages).	Therefore, it's okay
 * to discard any pending LocalInvalidMsgs, since these will be redundant
 * with the global list.
 *
 * If not isCommit, we are aborting, and must locally process the messages
 * in our RollbackMsgs list.  No messages need be sent to other backends,
 * since they'll not have seen our changed tuples anyway.
 *
 * In any case, reset the various lists to empty.  We need not physically
 * free memory here, since TopTransactionContext is about to be emptied
 * anyway.
 *
 * Note:
 *		This should be called as the last step in processing a transaction.
 */
void
AtEOXactInvalidationMessages(bool isCommit)
{
	if (isCommit)
	{
		ProcessInvalidationMessages(&GlobalInvalidMsgs,
									SendSharedInvalidMessage);
	}
	else
	{
		ProcessInvalidationMessages(&RollbackMsgs,
									LocalExecuteInvalidationMessage);
	}

	DiscardInvalidationMessages(&GlobalInvalidMsgs, false);
	DiscardInvalidationMessages(&LocalInvalidMsgs, false);
	DiscardInvalidationMessages(&RollbackMsgs, false);
}

/*
 * CommandEndInvalidationMessages
 *		Process queued-up invalidation messages at end of one command
 *		in a transaction.
 *
 * Here, we send no messages to the shared queue, since we don't know yet if
 * we will commit.	But we do need to locally process the LocalInvalidMsgs
 * list, so as to flush our caches of any tuples we have outdated in the
 * current command.
 *
 * The isCommit = false case is not currently used, but may someday be
 * needed to support rollback to a savepoint within a transaction.
 * (I suspect it needs more work first --- tgl.)
 *
 * Note:
 *		This should be called during CommandCounterIncrement(),
 *		after we have advanced the command ID.
 */
void
CommandEndInvalidationMessages(bool isCommit)
{
	if (isCommit)
	{
		ProcessInvalidationMessages(&LocalInvalidMsgs,
									LocalExecuteInvalidationMessage);
	}
	else
	{
		ProcessInvalidationMessages(&RollbackMsgs,
									LocalExecuteInvalidationMessage);
	}

	/*
	 * LocalInvalidMsgs list is not interesting anymore, so flush it (for
	 * real).  Do *not* clear GlobalInvalidMsgs or RollbackMsgs.
	 */
	DiscardInvalidationMessages(&LocalInvalidMsgs, true);
}

/*
 * RelationInvalidateHeapTuple
 *		Register the given tuple for invalidation at end of command
 *		(ie, current command is outdating this tuple).
 */
void
RelationInvalidateHeapTuple(Relation relation, HeapTuple tuple)
{
	PrepareForTupleInvalidation(relation, tuple,
								RegisterCatcacheInvalidation,
								RegisterRelcacheInvalidation);
}

/*
 * RelationMark4RollbackHeapTuple
 *		Register the given tuple for invalidation in case of abort
 *		(ie, current command is creating this tuple).
 */
void
RelationMark4RollbackHeapTuple(Relation relation, HeapTuple tuple)
{
	PrepareForTupleInvalidation(relation, tuple,
								RegisterCatcacheRollback,
								RegisterRelcacheRollback);
}