*** src/backend/access/transam/multixact.c.orig	Fri Oct 14 22:59:37 2005
--- src/backend/access/transam/multixact.c	Thu Oct 27 18:59:20 2005
***************
*** 110,115 ****
--- 110,118 ----
  	/* next-to-be-assigned offset */
  	MultiXactOffset nextOffset;
  
+ 	/* flag to handle a corner case at MultiXactOffset wraparound */
+ 	bool		creatingOffsetZero;
+ 
  	/* the Offset SLRU area was last truncated at this MultiXactId */
  	MultiXactId lastTruncationPoint;
  
***************
*** 631,637 ****
  	}
  
  	/*
! 	 * OK, assign the MXID and offsets range to use
  	 */
  	multi = GetNewMultiXactId(nxids, &offset);
  
--- 634,649 ----
  	}
  
  	/*
! 	 * 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 ****
--- 680,701 ----
  	/* Now enter the information into the OFFSETs and MEMBERs logs */
  	RecordNewMultiXact(multi, offset, nxids, xids);
  
+ 	/*
+ 	 * If we created a multixact with starting offset zero, clear the flag
+ 	 * that says such an operation is in progress.  Since it's only a bool,
+ 	 * we probably don't really need to take the lock here, but let's be
+ 	 * safe.
+ 	 */
+ 	if (offset == 0)
+ 	{
+ 		LWLockAcquire(MultiXactGenLock, LW_EXCLUSIVE);
+ 		MultiXactState->creatingOffsetZero = false;
+ 		LWLockRelease(MultiXactGenLock);
+ 	}
+ 
+ 	/* Done with critical section */
+ 	END_CRIT_SECTION();
+ 
  	/* Store the new MultiXactId in the local cache, too */
  	mXactCachePut(multi, nxids, xids);
  
***************
*** 793,801 ****
--- 821,835 ----
  
  	/*
  	 * Reserve the members space.  Same considerations as above.
+ 	 *
+ 	 * Also, set creatingOffsetZero if we return zero as the starting offset.
+ 	 * See GetMultiXactIdMembers() for motivation.
  	 */
  	*offset = MultiXactState->nextOffset;
  
+ 	if (*offset == 0)
+ 		MultiXactState->creatingOffsetZero = true;
+ 
  	ExtendMultiXactMember(*offset, nxids);
  
  	MultiXactState->nextOffset += nxids;
***************
*** 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);
--- 902,951 ----
  		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 two 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).
! 	 * So, if we read zero as the next multixact offset, we have to treat
! 	 * it with suspicion.  It could be valid, though.  We deal with this
! 	 * ambiguity by requiring processes that are creating a multixact with
! 	 * starting offset zero to set the creatingOffsetZero flag in the shared
! 	 * data structure; we sleep until we see that cleared before trusting
! 	 * a zero offset.  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);
***************
*** 900,918 ****
  	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;
--- 957,976 ----
  	offset = *offptr;
  
  	/*
! 	 * 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);
--- 985,1008 ----
  
  		offptr = (MultiXactOffset *) MultiXactOffsetCtl->shared->page_buffer[slotno];
  		offptr += entryno;
! 		nextMXOffset = *offptr;
! 
! 		/*
! 		 * Corner case 2: can we trust the next offset?
! 		 *
! 		 * We don't bother to lock the shared state; reading a boolean should
! 		 * be safe enough.
! 		 */
! 		if (nextMXOffset == 0 &&
! 			MultiXactState->creatingOffsetZero)
! 		{
! 			/* No, so sleep and retry */
! 			LWLockRelease(MultiXactOffsetControlLock);
! 			pg_usleep(1000L);
! 			goto retry;
! 		}
! 
! 		length = nextMXOffset - offset;
  	}
  
  	LWLockRelease(MultiXactOffsetControlLock);
***************
*** 1360,1365 ****
--- 1435,1445 ----
  	}
  
  	LWLockRelease(MultiXactMemberControlLock);
+ 
+ 	/*
+ 	 * Initialize other shared state.
+ 	 */
+ 	MultiXactState->creatingOffsetZero = false;
  
  	/*
  	 * Initialize lastTruncationPoint to invalid, ensuring that the first
