*** src/backend/access/transam/multixact.c.orig	Fri Oct 14 22:59:37 2005
--- src/backend/access/transam/multixact.c	Fri Oct 28 12:00:58 2005
***************
*** 631,637 ****
  	}
  
  	/*
! 	 * OK, assign the MXID and offsets range to use
  	 */
  	multi = GetNewMultiXactId(nxids, &offset);
  
--- 631,646 ----
  	}
  
  	/*
! 	 * Critical section from here until we've written the data; we don't
! 	 * want to error out with a partly written MultiXact structure.
! 	 * (In particular, failing to write our start offset after advancing
! 	 * nextMXact would effectively corrupt the previous MultiXact.)
! 	 */
! 	START_CRIT_SECTION();
! 
! 	/*
! 	 * Assign the MXID and offsets range to use, and make sure there is
! 	 * space in the OFFSETs and MEMBERs files.
  	 */
  	multi = GetNewMultiXactId(nxids, &offset);
  
***************
*** 668,673 ****
--- 677,685 ----
  	/* Now enter the information into the OFFSETs and MEMBERs logs */
  	RecordNewMultiXact(multi, offset, nxids, xids);
  
+ 	/* Done with critical section */
+ 	END_CRIT_SECTION();
+ 
  	/* Store the new MultiXactId in the local cache, too */
  	mXactCachePut(multi, nxids, xids);
  
***************
*** 761,766 ****
--- 773,779 ----
  GetNewMultiXactId(int nxids, MultiXactOffset *offset)
  {
  	MultiXactId result;
+ 	MultiXactOffset nextOffset;
  
  	debug_elog3(DEBUG2, "GetNew: for %d xids", nxids);
  
***************
*** 784,802 ****
  	 * Advance counter.  As in GetNewTransactionId(), this must not happen
  	 * until after ExtendMultiXactOffset has succeeded!
  	 *
! 	 * We don't care about MultiXactId wraparound here; it will be handled by the
! 	 * next iteration.	But note that nextMXact may be InvalidMultiXactId
  	 * after this routine exits, so anyone else looking at the variable must
  	 * be prepared to deal with that.
  	 */
  	(MultiXactState->nextMXact)++;
  
  	/*
! 	 * Reserve the members space.  Same considerations as above.
  	 */
! 	*offset = MultiXactState->nextOffset;
  
! 	ExtendMultiXactMember(*offset, nxids);
  
  	MultiXactState->nextOffset += nxids;
  
--- 797,824 ----
  	 * Advance counter.  As in GetNewTransactionId(), this must not happen
  	 * until after ExtendMultiXactOffset has succeeded!
  	 *
! 	 * We don't care about MultiXactId wraparound here; it will be handled by
! 	 * the next iteration.	But note that nextMXact may be InvalidMultiXactId
  	 * after this routine exits, so anyone else looking at the variable must
  	 * be prepared to deal with that.
  	 */
  	(MultiXactState->nextMXact)++;
  
  	/*
! 	 * Reserve the members space.  Same considerations as above.  Also, be
! 	 * careful not to return zero as the starting offset for any multixact.
! 	 * See GetMultiXactIdMembers() for motivation.
  	 */
! 	nextOffset = MultiXactState->nextOffset;
! 	if (nextOffset == 0)
! 	{
! 		*offset = 1;
! 		nxids++;				/* allocate member slot 0 too */
! 	}
! 	else
! 		*offset = nextOffset;
  
! 	ExtendMultiXactMember(nextOffset, nxids);
  
  	MultiXactState->nextOffset += nxids;
  
***************
*** 824,829 ****
--- 846,852 ----
  	MultiXactOffset *offptr;
  	MultiXactOffset offset;
  	int			length;
+ 	int			truelength;
  	int			i;
  	MultiXactId nextMXact;
  	MultiXactId tmpMXact;
***************
*** 849,861 ****
  	/*
  	 * We check known limits on MultiXact before resorting to the SLRU area.
  	 *
! 	 * An ID older than our OldestVisibleMXactId[] entry can't possibly still be
! 	 * running, and we'd run the risk of trying to read already-truncated SLRU
! 	 * data if we did try to examine it.
  	 *
! 	 * Conversely, an ID >= nextMXact shouldn't ever be seen here; if it is seen,
! 	 * it implies undetected ID wraparound has occurred.  We just silently
! 	 * assume that such an ID is no longer running.
  	 *
  	 * Shared lock is enough here since we aren't modifying any global state.
  	 * Also, we can examine our own OldestVisibleMXactId without the lock,
--- 872,884 ----
  	/*
  	 * We check known limits on MultiXact before resorting to the SLRU area.
  	 *
! 	 * An ID older than our OldestVisibleMXactId[] entry can't possibly still
! 	 * be running, and we'd run the risk of trying to read already-truncated
! 	 * SLRU data if we did try to examine it.
  	 *
! 	 * Conversely, an ID >= nextMXact shouldn't ever be seen here; if it is
! 	 * seen, it implies undetected ID wraparound has occurred.  We just
! 	 * silently assume that such an ID is no longer running.
  	 *
  	 * Shared lock is enough here since we aren't modifying any global state.
  	 * Also, we can examine our own OldestVisibleMXactId without the lock,
***************
*** 868,894 ****
  		return -1;
  	}
  
  	LWLockAcquire(MultiXactGenLock, LW_SHARED);
  
! 	if (!MultiXactIdPrecedes(multi, MultiXactState->nextMXact))
  	{
- 		LWLockRelease(MultiXactGenLock);
  		debug_elog2(DEBUG2, "GetMembers: it's too new!");
  		*xids = NULL;
  		return -1;
  	}
  
  	/*
! 	 * Before releasing the lock, save the current counter values, because the
! 	 * target MultiXactId may be just one less than nextMXact.	We will need
! 	 * to use nextOffset as the endpoint if so.
  	 */
! 	nextMXact = MultiXactState->nextMXact;
! 	nextOffset = MultiXactState->nextOffset;
! 
! 	LWLockRelease(MultiXactGenLock);
! 
! 	/* Get the offset at which we need to start reading MultiXactMembers */
  	LWLockAcquire(MultiXactOffsetControlLock, LW_EXCLUSIVE);
  
  	pageno = MultiXactIdToOffsetPage(multi);
--- 891,948 ----
  		return -1;
  	}
  
+ 	/*
+ 	 * Acquire the shared lock just long enough to grab the current counter
+ 	 * values.  We may need both nextMXact and nextOffset; see below.
+ 	 */
  	LWLockAcquire(MultiXactGenLock, LW_SHARED);
  
! 	nextMXact = MultiXactState->nextMXact;
! 	nextOffset = MultiXactState->nextOffset;
! 
! 	LWLockRelease(MultiXactGenLock);
! 
! 	if (!MultiXactIdPrecedes(multi, nextMXact))
  	{
  		debug_elog2(DEBUG2, "GetMembers: it's too new!");
  		*xids = NULL;
  		return -1;
  	}
  
  	/*
! 	 * Find out the offset at which we need to start reading MultiXactMembers
! 	 * and the number of members in the multixact.  We determine the latter
! 	 * as the difference between this multixact's starting offset and the
! 	 * next one's.  However, there are some corner cases to worry about:
! 	 *
! 	 * 1. This multixact may be the latest one created, in which case there
! 	 * is no next one to look at.  In this case the nextOffset value we just
! 	 * saved is the correct endpoint.
! 	 *
! 	 * 2. The next multixact may still be in process of being filled in:
! 	 * that is, another process may have done GetNewMultiXactId but not yet
! 	 * written the offset entry for that ID.  In that scenario, it is
! 	 * guaranteed that the offset entry for that multixact exists (because
! 	 * GetNewMultiXactId won't release MultiXactGenLock until it does)
! 	 * but contains zero (because we are careful to pre-zero offset pages).
! 	 * Because GetNewMultiXactId will never return zero as the starting offset
! 	 * for a multixact, when we read zero as the next multixact's offset, we
! 	 * know we have this case.  We sleep for a bit and try again.
! 	 *
! 	 * 3. Because GetNewMultiXactId increments offset zero to offset one
! 	 * to handle case #2, there is an ambiguity near the point of offset
! 	 * wraparound.  If we see next multixact's offset is one, is that our
! 	 * multixact's actual endpoint, or did it end at zero with a subsequent
! 	 * increment?  We handle this using the knowledge that if the zero'th
! 	 * member slot wasn't filled, it'll contain zero, and zero isn't a valid
! 	 * transaction ID so it can't be a multixact member.  Therefore, if we
! 	 * read a zero from the members array, just ignore it.
! 	 *
! 	 * This is all pretty messy, but the mess occurs only in infrequent corner
! 	 * cases, so it seems better than holding the MultiXactGenLock for a long
! 	 * time on every multixact creation.
  	 */
! retry:
  	LWLockAcquire(MultiXactOffsetControlLock, LW_EXCLUSIVE);
  
  	pageno = MultiXactIdToOffsetPage(multi);
***************
*** 899,918 ****
  	offptr += entryno;
  	offset = *offptr;
  
  	/*
! 	 * How many members do we need to read?  If we are at the end of the
! 	 * assigned MultiXactIds, use the offset just saved above.	Else we need
! 	 * to check the MultiXactId following ours.
! 	 *
! 	 * Use the same increment rule as GetNewMultiXactId(), that is, don't handle
! 	 * wraparound explicitly until needed.
  	 */
  	tmpMXact = multi + 1;
  
  	if (nextMXact == tmpMXact)
  		length = nextOffset - offset;
  	else
  	{
  		/* handle wraparound if needed */
  		if (tmpMXact < FirstMultiXactId)
  			tmpMXact = FirstMultiXactId;
--- 953,975 ----
  	offptr += entryno;
  	offset = *offptr;
  
+ 	Assert(offset != 0);
+ 
  	/*
! 	 * Use the same increment rule as GetNewMultiXactId(), that is, don't
! 	 * handle wraparound explicitly until needed.
  	 */
  	tmpMXact = multi + 1;
  
  	if (nextMXact == tmpMXact)
+ 	{
+ 		/* Corner case 1: there is no next multixact */
  		length = nextOffset - offset;
+ 	}
  	else
  	{
+ 		MultiXactOffset nextMXOffset;
+ 
  		/* handle wraparound if needed */
  		if (tmpMXact < FirstMultiXactId)
  			tmpMXact = FirstMultiXactId;
***************
*** 927,933 ****
  
  		offptr = (MultiXactOffset *) MultiXactOffsetCtl->shared->page_buffer[slotno];
  		offptr += entryno;
! 		length = *offptr - offset;
  	}
  
  	LWLockRelease(MultiXactOffsetControlLock);
--- 984,1000 ----
  
  		offptr = (MultiXactOffset *) MultiXactOffsetCtl->shared->page_buffer[slotno];
  		offptr += entryno;
! 		nextMXOffset = *offptr;
! 
! 		if (nextMXOffset == 0)
! 		{
! 			/* Corner case 2: next multixact is still being filled in */
! 			LWLockRelease(MultiXactOffsetControlLock);
! 			pg_usleep(1000L);
! 			goto retry;
! 		}
! 
! 		length = nextMXOffset - offset;
  	}
  
  	LWLockRelease(MultiXactOffsetControlLock);
***************
*** 938,943 ****
--- 1005,1011 ----
  	/* Now get the members themselves. */
  	LWLockAcquire(MultiXactMemberControlLock, LW_EXCLUSIVE);
  
+ 	truelength = 0;
  	prev_pageno = -1;
  	for (i = 0; i < length; i++, offset++)
  	{
***************
*** 956,962 ****
  			MultiXactMemberCtl->shared->page_buffer[slotno];
  		xactptr += entryno;
  
! 		ptr[i] = *xactptr;
  	}
  
  	LWLockRelease(MultiXactMemberControlLock);
--- 1024,1037 ----
  			MultiXactMemberCtl->shared->page_buffer[slotno];
  		xactptr += entryno;
  
! 		if (!TransactionIdIsValid(*xactptr))
! 		{
! 			/* Corner case 3: we must be looking at unused slot zero */
! 			Assert(offset == 0);
! 			continue;
! 		}
! 
! 		ptr[truelength++] = *xactptr;
  	}
  
  	LWLockRelease(MultiXactMemberControlLock);
***************
*** 964,974 ****
  	/*
  	 * Copy the result into the local cache.
  	 */
! 	mXactCachePut(multi, length, ptr);
  
  	debug_elog3(DEBUG2, "GetMembers: no cache for %s",
! 				mxid_to_string(multi, length, ptr));
! 	return length;
  }
  
  /*
--- 1039,1049 ----
  	/*
  	 * Copy the result into the local cache.
  	 */
! 	mXactCachePut(multi, truelength, ptr);
  
  	debug_elog3(DEBUG2, "GetMembers: no cache for %s",
! 				mxid_to_string(multi, truelength, ptr));
! 	return truelength;
  }
  
  /*
