Relation extension scalability

>From fc095897a6f4207d384559a095f80a36cf49648c Mon Sep 17 00:00:00 2001
From: Andres Freund <andres@anarazel.de>
Date: Sun, 29 Mar 2015 20:55:32 +0200
Subject: [PATCH] WIP: Saner heap extension.

---
 src/backend/access/heap/hio.c       |  86 ++++----
 src/backend/commands/vacuumlazy.c   |  39 ++--
 src/backend/storage/buffer/bufmgr.c | 377 ++++++++++++++++++++++++++----------
 src/backend/storage/smgr/md.c       |  62 ++++++
 src/backend/storage/smgr/smgr.c     |  20 +-
 src/include/storage/buf_internals.h |   1 +
 src/include/storage/bufmgr.h        |   1 +
 src/include/storage/smgr.h          |   7 +-
 8 files changed, 417 insertions(+), 176 deletions(-)

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 6db73bf..b47f9fe 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -15,6 +15,8 @@
 
 #include "postgres.h"
 
+#include "miscadmin.h"
+
 #include "access/heapam.h"
 #include "access/hio.h"
 #include "access/htup_details.h"
@@ -237,7 +239,6 @@ RelationGetBufferForTuple(Relation relation, Size len,
 				saveFreeSpace;
 	BlockNumber targetBlock,
 				otherBlock;
-	bool		needLock;
 
 	len = MAXALIGN(len);		/* be conservative */
 
@@ -433,63 +434,50 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	/*
 	 * Have to extend the relation.
 	 *
-	 * We have to use a lock to ensure no one else is extending the rel at the
-	 * same time, else we will both try to initialize the same new page.  We
-	 * can skip locking for new or temp relations, however, since no one else
-	 * could be accessing them.
+	 * To avoid, as it used to be the case, holding the extension lock during
+	 * victim buffer search for the new buffer, we extend the relation here
+	 * instead of relying on bufmgr.c. We still have to hold the extension
+	 * lock to prevent a race between two backends initializing the same page.
 	 */
-	needLock = !RELATION_IS_LOCAL(relation);
-
-	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
+	while(true)
+	{
+		buffer = ExtendRelation(relation, MAIN_FORKNUM, bistate->strategy);
 
-	/*
-	 * XXX This does an lseek - rather expensive - but at the moment it is the
-	 * only way to accurately determine how many blocks are in a relation.  Is
-	 * it worth keeping an accurate file length in shared memory someplace,
-	 * rather than relying on the kernel to do it for us?
-	 */
-	buffer = ReadBufferBI(relation, P_NEW, bistate);
+		if (otherBuffer != InvalidBuffer)
+			LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
 
-	/*
-	 * We can be certain that locking the otherBuffer first is OK, since it
-	 * must have a lower page number.
-	 */
-	if (otherBuffer != InvalidBuffer)
-		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+		/*
+		 * Now acquire lock on the new page.
+		 */
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
 
-	/*
-	 * Now acquire lock on the new page.
-	 */
-	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+		page = BufferGetPage(buffer);
 
-	/*
-	 * Release the file-extension lock; it's now OK for someone else to extend
-	 * the relation some more.  Note that we cannot release this lock before
-	 * we have buffer lock on the new page, or we risk a race condition
-	 * against vacuumlazy.c --- see comments therein.
-	 */
-	if (needLock)
-		UnlockRelationForExtension(relation, ExclusiveLock);
+		/*
+		 * While unlikely, it's possible that another backend managed to
+		 * initialize the page and use up the free space till we got the
+		 * exclusive lock. That'd require the page to be vacuumed (to be put
+		 * on the free space list) and then be used; possible but fairly
+		 * unlikely in practice. If it happens and there's not enough space,
+		 * just retry.
+		 */
+		if (PageIsNew(page))
+		{
+			PageInit(page, BLCKSZ, 0);
 
-	/*
-	 * We need to initialize the empty new page.  Double-check that it really
-	 * is empty (this should never happen, but if it does we don't want to
-	 * risk wiping out valid data).
-	 */
-	page = BufferGetPage(buffer);
+			Assert(len <= PageGetHeapFreeSpace(page));
+			break;
+		}
+		else if (len <= PageGetHeapFreeSpace(page))
+			break;
 
-	if (!PageIsNew(page))
-		elog(ERROR, "page %u of relation \"%s\" should be empty but is not",
-			 BufferGetBlockNumber(buffer),
-			 RelationGetRelationName(relation));
+		if (otherBuffer != InvalidBuffer)
+			LockBuffer(otherBuffer, BUFFER_LOCK_UNLOCK);
 
-	PageInit(page, BufferGetPageSize(buffer), 0);
+		LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
+		ReleaseBuffer(buffer);
 
-	if (len > PageGetHeapFreeSpace(page))
-	{
-		/* We should not get here given the test at the top */
-		elog(PANIC, "tuple is too big: size %zu", len);
+		CHECK_FOR_INTERRUPTS();
 	}
 
 	/*
diff --git a/src/backend/commands/vacuumlazy.c b/src/backend/commands/vacuumlazy.c
index a01cfb4..896731c 100644
--- a/src/backend/commands/vacuumlazy.c
+++ b/src/backend/commands/vacuumlazy.c
@@ -674,35 +674,18 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 			/*
 			 * An all-zeroes page could be left over if a backend extends the
 			 * relation but crashes before initializing the page. Reclaim such
-			 * pages for use.
-			 *
-			 * We have to be careful here because we could be looking at a
-			 * page that someone has just added to the relation and not yet
-			 * been able to initialize (see RelationGetBufferForTuple). To
-			 * protect against that, release the buffer lock, grab the
-			 * relation extension lock momentarily, and re-lock the buffer. If
-			 * the page is still uninitialized by then, it must be left over
-			 * from a crashed backend, and we can initialize it.
-			 *
-			 * We don't really need the relation lock when this is a new or
-			 * temp relation, but it's probably not worth the code space to
-			 * check that, since this surely isn't a critical path.
-			 *
-			 * Note: the comparable code in vacuum.c need not worry because
-			 * it's got exclusive lock on the whole relation.
+			 * pages for use.  It is also possible that we're looking at a
+			 * page that has just added but not yet initialized (see
+			 * RelationGetBufferForTuple). In that case we just initialize the
+			 * page here. That means the page will end up in the free space
+			 * map a little earlier, but that seems fine.
 			 */
-			LockBuffer(buf, BUFFER_LOCK_UNLOCK);
-			LockRelationForExtension(onerel, ExclusiveLock);
-			UnlockRelationForExtension(onerel, ExclusiveLock);
-			LockBufferForCleanup(buf);
-			if (PageIsNew(page))
-			{
-				ereport(WARNING,
-				(errmsg("relation \"%s\" page %u is uninitialized --- fixing",
-						relname, blkno)));
-				PageInit(page, BufferGetPageSize(buf), 0);
-				empty_pages++;
-			}
+			ereport(DEBUG2,
+					(errmsg("relation \"%s\" page %u is uninitialized --- fixing",
+							relname, blkno)));
+			PageInit(page, BufferGetPageSize(buf), 0);
+			empty_pages++;
+
 			freespace = PageGetHeapFreeSpace(page);
 			MarkBufferDirty(buf);
 			UnlockReleaseBuffer(buf);
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index e4b25587..4613666 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -392,6 +392,7 @@ static Buffer ReadBuffer_common(SMgrRelation reln, char relpersistence,
 				  ForkNumber forkNum, BlockNumber blockNum,
 				  ReadBufferMode mode, BufferAccessStrategy strategy,
 				  bool *hit);
+static volatile BufferDesc *GetVictimBuffer(BufferAccessStrategy strategy, BufFlags *oldFlags);
 static bool PinBuffer(volatile BufferDesc *buf, BufferAccessStrategy strategy);
 static void PinBuffer_Locked(volatile BufferDesc *buf);
 static void UnpinBuffer(volatile BufferDesc *buf, bool fixOwner);
@@ -483,6 +484,176 @@ PrefetchBuffer(Relation reln, ForkNumber forkNum, BlockNumber blockNum)
 #endif   /* USE_PREFETCH */
 }
 
+Buffer
+ExtendRelation(Relation reln, ForkNumber forkNum, BufferAccessStrategy strategy)
+{
+	BlockNumber	blockno;
+	Buffer		buf_id;
+	volatile BufferDesc *buf;
+	BufFlags	oldFlags;
+	Block		bufBlock;
+	bool		isLocalBuf = RelationUsesLocalBuffers(reln);
+	int			readblocks;
+
+	BufferTag	oldTag;			/* previous identity of selected buffer */
+	uint32		oldHash;		/* hash value for oldTag */
+	LWLock	   *oldPartitionLock;		/* buffer partition lock for it */
+
+	BufferTag	newTag;
+	uint32		newHash;
+	LWLock	   *newPartitionLock;
+
+	/* FIXME: This obviously isn't acceptable for integration */
+	if (isLocalBuf)
+	{
+		return ReadBufferExtended(reln, forkNum, P_NEW, RBM_NORMAL, strategy);
+	}
+
+	/* Open it at the smgr level if not already done */
+	RelationOpenSmgr(reln);
+
+	/* Make sure we will have room to remember the buffer pin */
+	ResourceOwnerEnlargeBuffers(CurrentResourceOwner);
+
+retry_victim:
+	/* we'll need a clean unassociated victim buffer */
+	while (true)
+	{
+		bool		gotIt = false;
+
+		/*
+		 * Returns a buffer that was unpinned and not dirty at the time of the
+		 * check.
+		 */
+		buf = GetVictimBuffer(strategy, &oldFlags);
+
+		if (oldFlags & BM_TAG_VALID)
+		{
+			oldTag = buf->tag;
+			oldHash = BufTableHashCode(&oldTag);
+			oldPartitionLock = BufMappingPartitionLock(oldHash);
+			LWLockAcquire(oldPartitionLock, LW_EXCLUSIVE);
+		}
+
+		LockBufHdr(buf);
+
+		/* somebody else might have re-pinned the buffer by now */
+		if (buf->refcount != 1 || (buf->flags & BM_DIRTY))
+		{
+			UnlockBufHdr(buf);
+		}
+		else
+		{
+			buf->flags &= ~(BM_TAG_VALID | BM_VALID | BM_DIRTY | BM_JUST_DIRTIED | BM_CHECKPOINT_NEEDED | BM_IO_ERROR | BM_PERMANENT | BM_NEW);
+
+			UnlockBufHdr(buf);
+
+			gotIt = true;
+
+			if (oldFlags & BM_TAG_VALID)
+				BufTableDelete(&oldTag, oldHash);
+		}
+
+		if (oldFlags & BM_TAG_VALID)
+			LWLockRelease(oldPartitionLock);
+
+		if (gotIt)
+			break;
+		else
+			UnpinBuffer(buf, true);
+	}
+
+	/*
+	 * At this state we have an empty victim buffer; pinned to prevent it from
+	 * being reused.
+	 */
+
+	/*
+	 * First try the current end of the relation. If a concurrent process has
+	 * acquired that, try the next one after that.
+	 */
+	blockno = smgrnblocks(reln->rd_smgr, forkNum);
+
+	while (true)
+	{
+		INIT_BUFFERTAG(newTag, reln->rd_smgr->smgr_rnode.node, forkNum, blockno);
+
+		newHash = BufTableHashCode(&newTag);
+		newPartitionLock = BufMappingPartitionLock(newHash);
+		LWLockAcquire(newPartitionLock, LW_EXCLUSIVE);
+
+		buf_id = BufTableInsert(&newTag, newHash, buf->buf_id);
+
+		if (buf_id >= 0)
+		{
+			/* somebody else got this block, try the next one */
+			LWLockRelease(newPartitionLock);
+			blockno++;
+			continue;
+		}
+
+		LockBufHdr(buf);
+
+		buf->tag = newTag;
+		if (reln->rd_rel->relpersistence == RELPERSISTENCE_PERMANENT)
+			buf->flags |= BM_NEW | BM_TAG_VALID | BM_PERMANENT;
+		else
+			buf->flags |= BM_NEW | BM_TAG_VALID;
+		buf->usage_count = 1;
+
+		UnlockBufHdr(buf);
+		LWLockRelease(newPartitionLock);
+
+		break;
+	}
+
+	/*
+	 * By here we made a entry into the buffer table, but haven't yet
+	 * read/written the page.  We can't just initialize the page, potentially
+	 * while we were busy with the above, another backend could have extended
+	 * the relation, written something, and the buffer could already have been
+	 * reused for something else.
+	 */
+
+	if (!StartBufferIO(buf, true))
+	{
+		/*
+		 * Somebody else is already using this block. Just try another one.
+		 */
+		UnpinBuffer(buf, true);
+		goto retry_victim;
+	}
+
+	/*
+	 * FIXME: if we die here we might have a problem: Everyone trying to read
+	 * this block will get a failure. Need to add checks for BM_NEW against
+	 * that. That's not really new to this code tho.
+	 */
+
+	bufBlock = isLocalBuf ? LocalBufHdrGetBlock(buf) : BufHdrGetBlock(buf);
+
+	readblocks = smgrtryread(reln->rd_smgr, forkNum, blockno, bufBlock);
+
+	if (readblocks != BLCKSZ)
+	{
+		MemSet((char *) bufBlock, 0, BLCKSZ);
+
+		smgrextend(reln->rd_smgr, forkNum, blockno, (char *) bufBlock, false);
+
+		/* Set BM_VALID, terminate IO, and wake up any waiters */
+		TerminateBufferIO(buf, false, BM_VALID);
+	}
+	else
+	{
+		/* Set BM_VALID, terminate IO, and wake up any waiters */
+		TerminateBufferIO(buf, false, BM_VALID);
+		UnpinBuffer(buf, true);
+
+		goto retry_victim;
+	}
+
+	return BufferDescriptorGetBuffer(buf);
+}
 
 /*
  * ReadBuffer -- a shorthand for ReadBufferExtended, for reading from main
@@ -847,6 +1018,112 @@ ReadBuffer_common(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
 	return BufferDescriptorGetBuffer(bufHdr);
 }
 
+static volatile BufferDesc *
+GetVictimBuffer(BufferAccessStrategy strategy, BufFlags *oldFlags)
+{
+	volatile BufferDesc *buf;
+
+	/*
+	 * Ensure, while the spinlock's not yet held, that there's a free refcount
+	 * entry.
+	 */
+	ReservePrivateRefCountEntry();
+
+retry:
+	/*
+	 * Select a victim buffer.  The buffer is returned with its header
+	 * spinlock still held!
+	 */
+	buf = StrategyGetBuffer(strategy);
+
+	Assert(buf->refcount == 0);
+
+	/* Must copy buffer flags while we still hold the spinlock */
+	*oldFlags = buf->flags;
+
+	/* Pin the buffer and then release the buffer spinlock */
+	PinBuffer_Locked(buf);
+
+	/*
+	 * If the buffer was dirty, try to write it out.  There is a race
+	 * condition here, in that someone might dirty it after we released it
+	 * above, or even while we are writing it out (since our share-lock
+	 * won't prevent hint-bit updates).  We will recheck the dirty bit
+	 * after re-locking the buffer header.
+	 */
+	if (*oldFlags & BM_DIRTY)
+	{
+		/*
+		 * We need a share-lock on the buffer contents to write it out
+		 * (else we might write invalid data, eg because someone else is
+		 * compacting the page contents while we write).  We must use a
+		 * conditional lock acquisition here to avoid deadlock.  Even
+		 * though the buffer was not pinned (and therefore surely not
+		 * locked) when StrategyGetBuffer returned it, someone else could
+		 * have pinned and exclusive-locked it by the time we get here. If
+		 * we try to get the lock unconditionally, we'd block waiting for
+		 * them; if they later block waiting for us, deadlock ensues.
+		 * (This has been observed to happen when two backends are both
+		 * trying to split btree index pages, and the second one just
+		 * happens to be trying to split the page the first one got from
+		 * StrategyGetBuffer.)
+		 */
+		if (LWLockConditionalAcquire(buf->content_lock, LW_SHARED))
+		{
+			/*
+			 * If using a nondefault strategy, and writing the buffer
+			 * would require a WAL flush, let the strategy decide whether
+			 * to go ahead and write/reuse the buffer or to choose another
+			 * victim.  We need lock to inspect the page LSN, so this
+			 * can't be done inside StrategyGetBuffer.
+			 */
+			if (strategy != NULL)
+			{
+				XLogRecPtr	lsn;
+
+				/* Read the LSN while holding buffer header lock */
+				LockBufHdr(buf);
+				lsn = BufferGetLSN(buf);
+				UnlockBufHdr(buf);
+
+				if (XLogNeedsFlush(lsn) &&
+					StrategyRejectBuffer(strategy, buf))
+				{
+					/* Drop lock/pin and loop around for another buffer */
+					LWLockRelease(buf->content_lock);
+					UnpinBuffer(buf, true);
+					goto retry;
+				}
+			}
+
+			/* OK, do the I/O */
+			TRACE_POSTGRESQL_BUFFER_WRITE_DIRTY_START(forkNum, blockNum,
+										   smgr->smgr_rnode.node.spcNode,
+											smgr->smgr_rnode.node.dbNode,
+										  smgr->smgr_rnode.node.relNode);
+
+			FlushBuffer(buf, NULL);
+			LWLockRelease(buf->content_lock);
+
+			TRACE_POSTGRESQL_BUFFER_WRITE_DIRTY_DONE(forkNum, blockNum,
+										   smgr->smgr_rnode.node.spcNode,
+											smgr->smgr_rnode.node.dbNode,
+										  smgr->smgr_rnode.node.relNode);
+		}
+		else
+		{
+			/*
+			 * Someone else has locked the buffer, so give it up and loop
+			 * back to get another one.
+			 */
+			UnpinBuffer(buf, true);
+			goto retry;
+		}
+	}
+
+	return buf;
+}
+
 /*
  * BufferAlloc -- subroutine for ReadBuffer.  Handles lookup of a shared
  *		buffer.  If no buffer exists already, selects a replacement
@@ -940,102 +1217,8 @@ BufferAlloc(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
 	/* Loop here in case we have to try another victim buffer */
 	for (;;)
 	{
-		/*
-		 * Ensure, while the spinlock's not yet held, that there's a free
-		 * refcount entry.
-		 */
-		ReservePrivateRefCountEntry();
-
-		/*
-		 * Select a victim buffer.  The buffer is returned with its header
-		 * spinlock still held!
-		 */
-		buf = StrategyGetBuffer(strategy);
-
-		Assert(buf->refcount == 0);
-
-		/* Must copy buffer flags while we still hold the spinlock */
-		oldFlags = buf->flags;
-
-		/* Pin the buffer and then release the buffer spinlock */
-		PinBuffer_Locked(buf);
-
-		/*
-		 * If the buffer was dirty, try to write it out.  There is a race
-		 * condition here, in that someone might dirty it after we released it
-		 * above, or even while we are writing it out (since our share-lock
-		 * won't prevent hint-bit updates).  We will recheck the dirty bit
-		 * after re-locking the buffer header.
-		 */
-		if (oldFlags & BM_DIRTY)
-		{
-			/*
-			 * We need a share-lock on the buffer contents to write it out
-			 * (else we might write invalid data, eg because someone else is
-			 * compacting the page contents while we write).  We must use a
-			 * conditional lock acquisition here to avoid deadlock.  Even
-			 * though the buffer was not pinned (and therefore surely not
-			 * locked) when StrategyGetBuffer returned it, someone else could
-			 * have pinned and exclusive-locked it by the time we get here. If
-			 * we try to get the lock unconditionally, we'd block waiting for
-			 * them; if they later block waiting for us, deadlock ensues.
-			 * (This has been observed to happen when two backends are both
-			 * trying to split btree index pages, and the second one just
-			 * happens to be trying to split the page the first one got from
-			 * StrategyGetBuffer.)
-			 */
-			if (LWLockConditionalAcquire(buf->content_lock, LW_SHARED))
-			{
-				/*
-				 * If using a nondefault strategy, and writing the buffer
-				 * would require a WAL flush, let the strategy decide whether
-				 * to go ahead and write/reuse the buffer or to choose another
-				 * victim.  We need lock to inspect the page LSN, so this
-				 * can't be done inside StrategyGetBuffer.
-				 */
-				if (strategy != NULL)
-				{
-					XLogRecPtr	lsn;
-
-					/* Read the LSN while holding buffer header lock */
-					LockBufHdr(buf);
-					lsn = BufferGetLSN(buf);
-					UnlockBufHdr(buf);
-
-					if (XLogNeedsFlush(lsn) &&
-						StrategyRejectBuffer(strategy, buf))
-					{
-						/* Drop lock/pin and loop around for another buffer */
-						LWLockRelease(buf->content_lock);
-						UnpinBuffer(buf, true);
-						continue;
-					}
-				}
-
-				/* OK, do the I/O */
-				TRACE_POSTGRESQL_BUFFER_WRITE_DIRTY_START(forkNum, blockNum,
-											   smgr->smgr_rnode.node.spcNode,
-												smgr->smgr_rnode.node.dbNode,
-											  smgr->smgr_rnode.node.relNode);
-
-				FlushBuffer(buf, NULL);
-				LWLockRelease(buf->content_lock);
-
-				TRACE_POSTGRESQL_BUFFER_WRITE_DIRTY_DONE(forkNum, blockNum,
-											   smgr->smgr_rnode.node.spcNode,
-												smgr->smgr_rnode.node.dbNode,
-											  smgr->smgr_rnode.node.relNode);
-			}
-			else
-			{
-				/*
-				 * Someone else has locked the buffer, so give it up and loop
-				 * back to get another one.
-				 */
-				UnpinBuffer(buf, true);
-				continue;
-			}
-		}
+		/* returns a nondirty buffer, with potentially valid contents */
+		buf = GetVictimBuffer(strategy, &oldFlags);
 
 		/*
 		 * To change the association of a valid buffer, we'll need to have
@@ -1171,7 +1354,7 @@ BufferAlloc(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
 	 * 1 so that the buffer can survive one clock-sweep pass.)
 	 */
 	buf->tag = newTag;
-	buf->flags &= ~(BM_VALID | BM_DIRTY | BM_JUST_DIRTIED | BM_CHECKPOINT_NEEDED | BM_IO_ERROR | BM_PERMANENT);
+	buf->flags &= ~(BM_VALID | BM_DIRTY | BM_JUST_DIRTIED | BM_CHECKPOINT_NEEDED | BM_IO_ERROR | BM_PERMANENT | BM_NEW);
 	if (relpersistence == RELPERSISTENCE_PERMANENT)
 		buf->flags |= BM_TAG_VALID | BM_PERMANENT;
 	else
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index 42a43bb..0038c91 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -729,6 +729,68 @@ mdread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 	}
 }
 
+
+/*
+ *	mdtryread() -- Read the specified block from a relation.
+ */
+int
+mdtryread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+	   char *buffer)
+{
+	off_t		seekpos;
+	int			nbytes;
+	MdfdVec    *v;
+
+	TRACE_POSTGRESQL_SMGR_MD_READ_START(forknum, blocknum,
+										reln->smgr_rnode.node.spcNode,
+										reln->smgr_rnode.node.dbNode,
+										reln->smgr_rnode.node.relNode,
+										reln->smgr_rnode.backend);
+
+	v = _mdfd_getseg(reln, forknum, blocknum, false, EXTENSION_RETURN_NULL);
+
+	/* would need another segment */
+	if (v == NULL)
+		return 0;
+
+	seekpos = (off_t) BLCKSZ *(blocknum % ((BlockNumber) RELSEG_SIZE));
+
+	Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+
+	if (FileSeek(v->mdfd_vfd, seekpos, SEEK_SET) != seekpos)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not seek to block %u in file \"%s\": %m",
+						blocknum, FilePathName(v->mdfd_vfd))));
+
+	nbytes = FileRead(v->mdfd_vfd, buffer, BLCKSZ);
+
+	TRACE_POSTGRESQL_SMGR_MD_READ_DONE(forknum, blocknum,
+									   reln->smgr_rnode.node.spcNode,
+									   reln->smgr_rnode.node.dbNode,
+									   reln->smgr_rnode.node.relNode,
+									   reln->smgr_rnode.backend,
+									   nbytes,
+									   BLCKSZ);
+
+	if (nbytes < 0)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read block %u in file \"%s\": %m",
+						blocknum, FilePathName(v->mdfd_vfd))));
+
+	if (nbytes > 0 && nbytes < BLCKSZ)
+	{
+		ereport(LOG,
+				(errcode(ERRCODE_DATA_CORRUPTED),
+				 errmsg("could not read block %u in file \"%s\": read only %d of %d bytes",
+						blocknum, FilePathName(v->mdfd_vfd),
+						nbytes, BLCKSZ)));
+	}
+
+	return nbytes;
+}
+
 /*
  *	mdwrite() -- Write the supplied block at the appropriate location.
  *
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index 244b4ea..f0e9a7b 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -51,6 +51,8 @@ typedef struct f_smgr
 											  BlockNumber blocknum);
 	void		(*smgr_read) (SMgrRelation reln, ForkNumber forknum,
 										  BlockNumber blocknum, char *buffer);
+	int			(*smgr_tryread) (SMgrRelation reln, ForkNumber forknum,
+										  BlockNumber blocknum, char *buffer);
 	void		(*smgr_write) (SMgrRelation reln, ForkNumber forknum,
 						 BlockNumber blocknum, char *buffer, bool skipFsync);
 	BlockNumber (*smgr_nblocks) (SMgrRelation reln, ForkNumber forknum);
@@ -66,7 +68,7 @@ typedef struct f_smgr
 static const f_smgr smgrsw[] = {
 	/* magnetic disk */
 	{mdinit, NULL, mdclose, mdcreate, mdexists, mdunlink, mdextend,
-		mdprefetch, mdread, mdwrite, mdnblocks, mdtruncate, mdimmedsync,
+		mdprefetch, mdread, mdtryread, mdwrite, mdnblocks, mdtruncate, mdimmedsync,
 		mdpreckpt, mdsync, mdpostckpt
 	}
 };
@@ -626,6 +628,22 @@ smgrread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 	(*(smgrsw[reln->smgr_which].smgr_read)) (reln, forknum, blocknum, buffer);
 }
 
+
+/*
+ *	smgtryrread() -- read a particular block from a relation into the supplied
+ *				  buffer.
+ *
+ *		This routine is called from the buffer manager in order to
+ *		instantiate pages in the shared buffer cache.  All storage managers
+ *		return pages in the format that POSTGRES expects.
+ */
+int
+smgrtryread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+		 char *buffer)
+{
+	return (*(smgrsw[reln->smgr_which].smgr_tryread)) (reln, forknum, blocknum, buffer);
+}
+
 /*
  *	smgrwrite() -- Write the supplied buffer out.
  *
diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index 521ee1c..5f961af 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -40,6 +40,7 @@
 #define BM_CHECKPOINT_NEEDED	(1 << 7)		/* must write for checkpoint */
 #define BM_PERMANENT			(1 << 8)		/* permanent relation (not
 												 * unlogged) */
+#define BM_NEW					(1 << 9)		/* Not guaranteed to exist on disk */
 
 typedef bits16 BufFlags;
 
diff --git a/src/include/storage/bufmgr.h b/src/include/storage/bufmgr.h
index ec0a254..b52591f 100644
--- a/src/include/storage/bufmgr.h
+++ b/src/include/storage/bufmgr.h
@@ -153,6 +153,7 @@ extern Buffer ReadBufferExtended(Relation reln, ForkNumber forkNum,
 extern Buffer ReadBufferWithoutRelcache(RelFileNode rnode,
 						  ForkNumber forkNum, BlockNumber blockNum,
 						  ReadBufferMode mode, BufferAccessStrategy strategy);
+extern Buffer ExtendRelation(Relation reln, ForkNumber forkNum, BufferAccessStrategy strategy);
 extern void ReleaseBuffer(Buffer buffer);
 extern void UnlockReleaseBuffer(Buffer buffer);
 extern void MarkBufferDirty(Buffer buffer);
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index 69a624f..07a331c 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -94,6 +94,8 @@ extern void smgrprefetch(SMgrRelation reln, ForkNumber forknum,
 			 BlockNumber blocknum);
 extern void smgrread(SMgrRelation reln, ForkNumber forknum,
 		 BlockNumber blocknum, char *buffer);
+extern int smgrtryread(SMgrRelation reln, ForkNumber forknum,
+		 BlockNumber blocknum, char *buffer);
 extern void smgrwrite(SMgrRelation reln, ForkNumber forknum,
 		  BlockNumber blocknum, char *buffer, bool skipFsync);
 extern BlockNumber smgrnblocks(SMgrRelation reln, ForkNumber forknum);
@@ -114,12 +116,15 @@ extern void mdclose(SMgrRelation reln, ForkNumber forknum);
 extern void mdcreate(SMgrRelation reln, ForkNumber forknum, bool isRedo);
 extern bool mdexists(SMgrRelation reln, ForkNumber forknum);
 extern void mdunlink(RelFileNodeBackend rnode, ForkNumber forknum, bool isRedo);
-extern void mdextend(SMgrRelation reln, ForkNumber forknum,
+extern void mdextend(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, char *buffer, bool skipFsync);
+extern void mdappend(SMgrRelation reln, ForkNumber forknum,
 		 BlockNumber blocknum, char *buffer, bool skipFsync);
 extern void mdprefetch(SMgrRelation reln, ForkNumber forknum,
 		   BlockNumber blocknum);
 extern void mdread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 	   char *buffer);
+extern int mdtryread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+	   char *buffer);
 extern void mdwrite(SMgrRelation reln, ForkNumber forknum,
 		BlockNumber blocknum, char *buffer, bool skipFsync);
 extern BlockNumber mdnblocks(SMgrRelation reln, ForkNumber forknum);
-- 
2.3.0.149.gf3f4077.dirty

>From fc095897a6f4207d384559a095f80a36cf49648c Mon Sep 17 00:00:00 2001
From: Andres Freund <andres@anarazel.de>
Date: Sun, 29 Mar 2015 20:55:32 +0200
Subject: [PATCH] WIP: Saner heap extension.

---
 src/backend/access/heap/hio.c       |  86 ++++----
 src/backend/commands/vacuumlazy.c   |  39 ++--
 src/backend/storage/buffer/bufmgr.c | 377 ++++++++++++++++++++++++++----------
 src/backend/storage/smgr/md.c       |  62 ++++++
 src/backend/storage/smgr/smgr.c     |  20 +-
 src/include/storage/buf_internals.h |   1 +
 src/include/storage/bufmgr.h        |   1 +
 src/include/storage/smgr.h          |   7 +-
 8 files changed, 417 insertions(+), 176 deletions(-)

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 6db73bf..b47f9fe 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -15,6 +15,8 @@
 
 #include "postgres.h"
 
+#include "miscadmin.h"
+
 #include "access/heapam.h"
 #include "access/hio.h"
 #include "access/htup_details.h"
@@ -237,7 +239,6 @@ RelationGetBufferForTuple(Relation relation, Size len,
 				saveFreeSpace;
 	BlockNumber targetBlock,
 				otherBlock;
-	bool		needLock;
 
 	len = MAXALIGN(len);		/* be conservative */
 
@@ -433,63 +434,50 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	/*
 	 * Have to extend the relation.
 	 *
-	 * We have to use a lock to ensure no one else is extending the rel at the
-	 * same time, else we will both try to initialize the same new page.  We
-	 * can skip locking for new or temp relations, however, since no one else
-	 * could be accessing them.
+	 * To avoid, as it used to be the case, holding the extension lock during
+	 * victim buffer search for the new buffer, we extend the relation here
+	 * instead of relying on bufmgr.c. We still have to hold the extension
+	 * lock to prevent a race between two backends initializing the same page.
 	 */
-	needLock = !RELATION_IS_LOCAL(relation);
-
-	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
+	while(true)
+	{
+		buffer = ExtendRelation(relation, MAIN_FORKNUM, bistate->strategy);
 
-	/*
-	 * XXX This does an lseek - rather expensive - but at the moment it is the
-	 * only way to accurately determine how many blocks are in a relation.  Is
-	 * it worth keeping an accurate file length in shared memory someplace,
-	 * rather than relying on the kernel to do it for us?
-	 */
-	buffer = ReadBufferBI(relation, P_NEW, bistate);
+		if (otherBuffer != InvalidBuffer)
+			LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
 
-	/*
-	 * We can be certain that locking the otherBuffer first is OK, since it
-	 * must have a lower page number.
-	 */
-	if (otherBuffer != InvalidBuffer)
-		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+		/*
+		 * Now acquire lock on the new page.
+		 */
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
 
-	/*
-	 * Now acquire lock on the new page.
-	 */
-	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+		page = BufferGetPage(buffer);
 
-	/*
-	 * Release the file-extension lock; it's now OK for someone else to extend
-	 * the relation some more.  Note that we cannot release this lock before
-	 * we have buffer lock on the new page, or we risk a race condition
-	 * against vacuumlazy.c --- see comments therein.
-	 */
-	if (needLock)
-		UnlockRelationForExtension(relation, ExclusiveLock);
+		/*
+		 * While unlikely, it's possible that another backend managed to
+		 * initialize the page and use up the free space till we got the
+		 * exclusive lock. That'd require the page to be vacuumed (to be put
+		 * on the free space list) and then be used; possible but fairly
+		 * unlikely in practice. If it happens and there's not enough space,
+		 * just retry.
+		 */
+		if (PageIsNew(page))
+		{
+			PageInit(page, BLCKSZ, 0);
 
-	/*
-	 * We need to initialize the empty new page.  Double-check that it really
-	 * is empty (this should never happen, but if it does we don't want to
-	 * risk wiping out valid data).
-	 */
-	page = BufferGetPage(buffer);
+			Assert(len <= PageGetHeapFreeSpace(page));
+			break;
+		}
+		else if (len <= PageGetHeapFreeSpace(page))
+			break;
 
-	if (!PageIsNew(page))
-		elog(ERROR, "page %u of relation \"%s\" should be empty but is not",
-			 BufferGetBlockNumber(buffer),
-			 RelationGetRelationName(relation));
+		if (otherBuffer != InvalidBuffer)
+			LockBuffer(otherBuffer, BUFFER_LOCK_UNLOCK);
 
-	PageInit(page, BufferGetPageSize(buffer), 0);
+		LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
+		ReleaseBuffer(buffer);
 
-	if (len > PageGetHeapFreeSpace(page))
-	{
-		/* We should not get here given the test at the top */
-		elog(PANIC, "tuple is too big: size %zu", len);
+		CHECK_FOR_INTERRUPTS();
 	}
 
 	/*
diff --git a/src/backend/commands/vacuumlazy.c b/src/backend/commands/vacuumlazy.c
index a01cfb4..896731c 100644
--- a/src/backend/commands/vacuumlazy.c
+++ b/src/backend/commands/vacuumlazy.c
@@ -674,35 +674,18 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 			/*
 			 * An all-zeroes page could be left over if a backend extends the
 			 * relation but crashes before initializing the page. Reclaim such
-			 * pages for use.
-			 *
-			 * We have to be careful here because we could be looking at a
-			 * page that someone has just added to the relation and not yet
-			 * been able to initialize (see RelationGetBufferForTuple). To
-			 * protect against that, release the buffer lock, grab the
-			 * relation extension lock momentarily, and re-lock the buffer. If
-			 * the page is still uninitialized by then, it must be left over
-			 * from a crashed backend, and we can initialize it.
-			 *
-			 * We don't really need the relation lock when this is a new or
-			 * temp relation, but it's probably not worth the code space to
-			 * check that, since this surely isn't a critical path.
-			 *
-			 * Note: the comparable code in vacuum.c need not worry because
-			 * it's got exclusive lock on the whole relation.
+			 * pages for use.  It is also possible that we're looking at a
+			 * page that has just added but not yet initialized (see
+			 * RelationGetBufferForTuple). In that case we just initialize the
+			 * page here. That means the page will end up in the free space
+			 * map a little earlier, but that seems fine.
 			 */
-			LockBuffer(buf, BUFFER_LOCK_UNLOCK);
-			LockRelationForExtension(onerel, ExclusiveLock);
-			UnlockRelationForExtension(onerel, ExclusiveLock);
-			LockBufferForCleanup(buf);
-			if (PageIsNew(page))
-			{
-				ereport(WARNING,
-				(errmsg("relation \"%s\" page %u is uninitialized --- fixing",
-						relname, blkno)));
-				PageInit(page, BufferGetPageSize(buf), 0);
-				empty_pages++;
-			}
+			ereport(DEBUG2,
+					(errmsg("relation \"%s\" page %u is uninitialized --- fixing",
+							relname, blkno)));
+			PageInit(page, BufferGetPageSize(buf), 0);
+			empty_pages++;
+
 			freespace = PageGetHeapFreeSpace(page);
 			MarkBufferDirty(buf);
 			UnlockReleaseBuffer(buf);
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index e4b25587..4613666 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -392,6 +392,7 @@ static Buffer ReadBuffer_common(SMgrRelation reln, char relpersistence,
 				  ForkNumber forkNum, BlockNumber blockNum,
 				  ReadBufferMode mode, BufferAccessStrategy strategy,
 				  bool *hit);
+static volatile BufferDesc *GetVictimBuffer(BufferAccessStrategy strategy, BufFlags *oldFlags);
 static bool PinBuffer(volatile BufferDesc *buf, BufferAccessStrategy strategy);
 static void PinBuffer_Locked(volatile BufferDesc *buf);
 static void UnpinBuffer(volatile BufferDesc *buf, bool fixOwner);
@@ -483,6 +484,176 @@ PrefetchBuffer(Relation reln, ForkNumber forkNum, BlockNumber blockNum)
 #endif   /* USE_PREFETCH */
 }
 
+Buffer
+ExtendRelation(Relation reln, ForkNumber forkNum, BufferAccessStrategy strategy)
+{
+	BlockNumber	blockno;
+	Buffer		buf_id;
+	volatile BufferDesc *buf;
+	BufFlags	oldFlags;
+	Block		bufBlock;
+	bool		isLocalBuf = RelationUsesLocalBuffers(reln);
+	int			readblocks;
+
+	BufferTag	oldTag;			/* previous identity of selected buffer */
+	uint32		oldHash;		/* hash value for oldTag */
+	LWLock	   *oldPartitionLock;		/* buffer partition lock for it */
+
+	BufferTag	newTag;
+	uint32		newHash;
+	LWLock	   *newPartitionLock;
+
+	/* FIXME: This obviously isn't acceptable for integration */
+	if (isLocalBuf)
+	{
+		return ReadBufferExtended(reln, forkNum, P_NEW, RBM_NORMAL, strategy);
+	}
+
+	/* Open it at the smgr level if not already done */
+	RelationOpenSmgr(reln);
+
+	/* Make sure we will have room to remember the buffer pin */
+	ResourceOwnerEnlargeBuffers(CurrentResourceOwner);
+
+retry_victim:
+	/* we'll need a clean unassociated victim buffer */
+	while (true)
+	{
+		bool		gotIt = false;
+
+		/*
+		 * Returns a buffer that was unpinned and not dirty at the time of the
+		 * check.
+		 */
+		buf = GetVictimBuffer(strategy, &oldFlags);
+
+		if (oldFlags & BM_TAG_VALID)
+		{
+			oldTag = buf->tag;
+			oldHash = BufTableHashCode(&oldTag);
+			oldPartitionLock = BufMappingPartitionLock(oldHash);
+			LWLockAcquire(oldPartitionLock, LW_EXCLUSIVE);
+		}
+
+		LockBufHdr(buf);
+
+		/* somebody else might have re-pinned the buffer by now */
+		if (buf->refcount != 1 || (buf->flags & BM_DIRTY))
+		{
+			UnlockBufHdr(buf);
+		}
+		else
+		{
+			buf->flags &= ~(BM_TAG_VALID | BM_VALID | BM_DIRTY | BM_JUST_DIRTIED | BM_CHECKPOINT_NEEDED | BM_IO_ERROR | BM_PERMANENT | BM_NEW);
+
+			UnlockBufHdr(buf);
+
+			gotIt = true;
+
+			if (oldFlags & BM_TAG_VALID)
+				BufTableDelete(&oldTag, oldHash);
+		}
+
+		if (oldFlags & BM_TAG_VALID)
+			LWLockRelease(oldPartitionLock);
+
+		if (gotIt)
+			break;
+		else
+			UnpinBuffer(buf, true);
+	}
+
+	/*
+	 * At this state we have an empty victim buffer; pinned to prevent it from
+	 * being reused.
+	 */
+
+	/*
+	 * First try the current end of the relation. If a concurrent process has
+	 * acquired that, try the next one after that.
+	 */
+	blockno = smgrnblocks(reln->rd_smgr, forkNum);
+
+	while (true)
+	{
+		INIT_BUFFERTAG(newTag, reln->rd_smgr->smgr_rnode.node, forkNum, blockno);
+
+		newHash = BufTableHashCode(&newTag);
+		newPartitionLock = BufMappingPartitionLock(newHash);
+		LWLockAcquire(newPartitionLock, LW_EXCLUSIVE);
+
+		buf_id = BufTableInsert(&newTag, newHash, buf->buf_id);
+
+		if (buf_id >= 0)
+		{
+			/* somebody else got this block, try the next one */
+			LWLockRelease(newPartitionLock);
+			blockno++;
+			continue;
+		}
+
+		LockBufHdr(buf);
+
+		buf->tag = newTag;
+		if (reln->rd_rel->relpersistence == RELPERSISTENCE_PERMANENT)
+			buf->flags |= BM_NEW | BM_TAG_VALID | BM_PERMANENT;
+		else
+			buf->flags |= BM_NEW | BM_TAG_VALID;
+		buf->usage_count = 1;
+
+		UnlockBufHdr(buf);
+		LWLockRelease(newPartitionLock);
+
+		break;
+	}
+
+	/*
+	 * By here we made a entry into the buffer table, but haven't yet
+	 * read/written the page.  We can't just initialize the page, potentially
+	 * while we were busy with the above, another backend could have extended
+	 * the relation, written something, and the buffer could already have been
+	 * reused for something else.
+	 */
+
+	if (!StartBufferIO(buf, true))
+	{
+		/*
+		 * Somebody else is already using this block. Just try another one.
+		 */
+		UnpinBuffer(buf, true);
+		goto retry_victim;
+	}
+
+	/*
+	 * FIXME: if we die here we might have a problem: Everyone trying to read
+	 * this block will get a failure. Need to add checks for BM_NEW against
+	 * that. That's not really new to this code tho.
+	 */
+
+	bufBlock = isLocalBuf ? LocalBufHdrGetBlock(buf) : BufHdrGetBlock(buf);
+
+	readblocks = smgrtryread(reln->rd_smgr, forkNum, blockno, bufBlock);
+
+	if (readblocks != BLCKSZ)
+	{
+		MemSet((char *) bufBlock, 0, BLCKSZ);
+
+		smgrextend(reln->rd_smgr, forkNum, blockno, (char *) bufBlock, false);
+
+		/* Set BM_VALID, terminate IO, and wake up any waiters */
+		TerminateBufferIO(buf, false, BM_VALID);
+	}
+	else
+	{
+		/* Set BM_VALID, terminate IO, and wake up any waiters */
+		TerminateBufferIO(buf, false, BM_VALID);
+		UnpinBuffer(buf, true);
+
+		goto retry_victim;
+	}
+
+	return BufferDescriptorGetBuffer(buf);
+}
 
 /*
  * ReadBuffer -- a shorthand for ReadBufferExtended, for reading from main
@@ -847,6 +1018,112 @@ ReadBuffer_common(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
 	return BufferDescriptorGetBuffer(bufHdr);
 }
 
+static volatile BufferDesc *
+GetVictimBuffer(BufferAccessStrategy strategy, BufFlags *oldFlags)
+{
+	volatile BufferDesc *buf;
+
+	/*
+	 * Ensure, while the spinlock's not yet held, that there's a free refcount
+	 * entry.
+	 */
+	ReservePrivateRefCountEntry();
+
+retry:
+	/*
+	 * Select a victim buffer.  The buffer is returned with its header
+	 * spinlock still held!
+	 */
+	buf = StrategyGetBuffer(strategy);
+
+	Assert(buf->refcount == 0);
+
+	/* Must copy buffer flags while we still hold the spinlock */
+	*oldFlags = buf->flags;
+
+	/* Pin the buffer and then release the buffer spinlock */
+	PinBuffer_Locked(buf);
+
+	/*
+	 * If the buffer was dirty, try to write it out.  There is a race
+	 * condition here, in that someone might dirty it after we released it
+	 * above, or even while we are writing it out (since our share-lock
+	 * won't prevent hint-bit updates).  We will recheck the dirty bit
+	 * after re-locking the buffer header.
+	 */
+	if (*oldFlags & BM_DIRTY)
+	{
+		/*
+		 * We need a share-lock on the buffer contents to write it out
+		 * (else we might write invalid data, eg because someone else is
+		 * compacting the page contents while we write).  We must use a
+		 * conditional lock acquisition here to avoid deadlock.  Even
+		 * though the buffer was not pinned (and therefore surely not
+		 * locked) when StrategyGetBuffer returned it, someone else could
+		 * have pinned and exclusive-locked it by the time we get here. If
+		 * we try to get the lock unconditionally, we'd block waiting for
+		 * them; if they later block waiting for us, deadlock ensues.
+		 * (This has been observed to happen when two backends are both
+		 * trying to split btree index pages, and the second one just
+		 * happens to be trying to split the page the first one got from
+		 * StrategyGetBuffer.)
+		 */
+		if (LWLockConditionalAcquire(buf->content_lock, LW_SHARED))
+		{
+			/*
+			 * If using a nondefault strategy, and writing the buffer
+			 * would require a WAL flush, let the strategy decide whether
+			 * to go ahead and write/reuse the buffer or to choose another
+			 * victim.  We need lock to inspect the page LSN, so this
+			 * can't be done inside StrategyGetBuffer.
+			 */
+			if (strategy != NULL)
+			{
+				XLogRecPtr	lsn;
+
+				/* Read the LSN while holding buffer header lock */
+				LockBufHdr(buf);
+				lsn = BufferGetLSN(buf);
+				UnlockBufHdr(buf);
+
+				if (XLogNeedsFlush(lsn) &&
+					StrategyRejectBuffer(strategy, buf))
+				{
+					/* Drop lock/pin and loop around for another buffer */
+					LWLockRelease(buf->content_lock);
+					UnpinBuffer(buf, true);
+					goto retry;
+				}
+			}
+
+			/* OK, do the I/O */
+			TRACE_POSTGRESQL_BUFFER_WRITE_DIRTY_START(forkNum, blockNum,
+										   smgr->smgr_rnode.node.spcNode,
+											smgr->smgr_rnode.node.dbNode,
+										  smgr->smgr_rnode.node.relNode);
+
+			FlushBuffer(buf, NULL);
+			LWLockRelease(buf->content_lock);
+
+			TRACE_POSTGRESQL_BUFFER_WRITE_DIRTY_DONE(forkNum, blockNum,
+										   smgr->smgr_rnode.node.spcNode,
+											smgr->smgr_rnode.node.dbNode,
+										  smgr->smgr_rnode.node.relNode);
+		}
+		else
+		{
+			/*
+			 * Someone else has locked the buffer, so give it up and loop
+			 * back to get another one.
+			 */
+			UnpinBuffer(buf, true);
+			goto retry;
+		}
+	}
+
+	return buf;
+}
+
 /*
  * BufferAlloc -- subroutine for ReadBuffer.  Handles lookup of a shared
  *		buffer.  If no buffer exists already, selects a replacement
@@ -940,102 +1217,8 @@ BufferAlloc(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
 	/* Loop here in case we have to try another victim buffer */
 	for (;;)
 	{
-		/*
-		 * Ensure, while the spinlock's not yet held, that there's a free
-		 * refcount entry.
-		 */
-		ReservePrivateRefCountEntry();
-
-		/*
-		 * Select a victim buffer.  The buffer is returned with its header
-		 * spinlock still held!
-		 */
-		buf = StrategyGetBuffer(strategy);
-
-		Assert(buf->refcount == 0);
-
-		/* Must copy buffer flags while we still hold the spinlock */
-		oldFlags = buf->flags;
-
-		/* Pin the buffer and then release the buffer spinlock */
-		PinBuffer_Locked(buf);
-
-		/*
-		 * If the buffer was dirty, try to write it out.  There is a race
-		 * condition here, in that someone might dirty it after we released it
-		 * above, or even while we are writing it out (since our share-lock
-		 * won't prevent hint-bit updates).  We will recheck the dirty bit
-		 * after re-locking the buffer header.
-		 */
-		if (oldFlags & BM_DIRTY)
-		{
-			/*
-			 * We need a share-lock on the buffer contents to write it out
-			 * (else we might write invalid data, eg because someone else is
-			 * compacting the page contents while we write).  We must use a
-			 * conditional lock acquisition here to avoid deadlock.  Even
-			 * though the buffer was not pinned (and therefore surely not
-			 * locked) when StrategyGetBuffer returned it, someone else could
-			 * have pinned and exclusive-locked it by the time we get here. If
-			 * we try to get the lock unconditionally, we'd block waiting for
-			 * them; if they later block waiting for us, deadlock ensues.
-			 * (This has been observed to happen when two backends are both
-			 * trying to split btree index pages, and the second one just
-			 * happens to be trying to split the page the first one got from
-			 * StrategyGetBuffer.)
-			 */
-			if (LWLockConditionalAcquire(buf->content_lock, LW_SHARED))
-			{
-				/*
-				 * If using a nondefault strategy, and writing the buffer
-				 * would require a WAL flush, let the strategy decide whether
-				 * to go ahead and write/reuse the buffer or to choose another
-				 * victim.  We need lock to inspect the page LSN, so this
-				 * can't be done inside StrategyGetBuffer.
-				 */
-				if (strategy != NULL)
-				{
-					XLogRecPtr	lsn;
-
-					/* Read the LSN while holding buffer header lock */
-					LockBufHdr(buf);
-					lsn = BufferGetLSN(buf);
-					UnlockBufHdr(buf);
-
-					if (XLogNeedsFlush(lsn) &&
-						StrategyRejectBuffer(strategy, buf))
-					{
-						/* Drop lock/pin and loop around for another buffer */
-						LWLockRelease(buf->content_lock);
-						UnpinBuffer(buf, true);
-						continue;
-					}
-				}
-
-				/* OK, do the I/O */
-				TRACE_POSTGRESQL_BUFFER_WRITE_DIRTY_START(forkNum, blockNum,
-											   smgr->smgr_rnode.node.spcNode,
-												smgr->smgr_rnode.node.dbNode,
-											  smgr->smgr_rnode.node.relNode);
-
-				FlushBuffer(buf, NULL);
-				LWLockRelease(buf->content_lock);
-
-				TRACE_POSTGRESQL_BUFFER_WRITE_DIRTY_DONE(forkNum, blockNum,
-											   smgr->smgr_rnode.node.spcNode,
-												smgr->smgr_rnode.node.dbNode,
-											  smgr->smgr_rnode.node.relNode);
-			}
-			else
-			{
-				/*
-				 * Someone else has locked the buffer, so give it up and loop
-				 * back to get another one.
-				 */
-				UnpinBuffer(buf, true);
-				continue;
-			}
-		}
+		/* returns a nondirty buffer, with potentially valid contents */
+		buf = GetVictimBuffer(strategy, &oldFlags);
 
 		/*
 		 * To change the association of a valid buffer, we'll need to have
@@ -1171,7 +1354,7 @@ BufferAlloc(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
 	 * 1 so that the buffer can survive one clock-sweep pass.)
 	 */
 	buf->tag = newTag;
-	buf->flags &= ~(BM_VALID | BM_DIRTY | BM_JUST_DIRTIED | BM_CHECKPOINT_NEEDED | BM_IO_ERROR | BM_PERMANENT);
+	buf->flags &= ~(BM_VALID | BM_DIRTY | BM_JUST_DIRTIED | BM_CHECKPOINT_NEEDED | BM_IO_ERROR | BM_PERMANENT | BM_NEW);
 	if (relpersistence == RELPERSISTENCE_PERMANENT)
 		buf->flags |= BM_TAG_VALID | BM_PERMANENT;
 	else
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index 42a43bb..0038c91 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -729,6 +729,68 @@ mdread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 	}
 }
 
+
+/*
+ *	mdtryread() -- Read the specified block from a relation.
+ */
+int
+mdtryread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+	   char *buffer)
+{
+	off_t		seekpos;
+	int			nbytes;
+	MdfdVec    *v;
+
+	TRACE_POSTGRESQL_SMGR_MD_READ_START(forknum, blocknum,
+										reln->smgr_rnode.node.spcNode,
+										reln->smgr_rnode.node.dbNode,
+										reln->smgr_rnode.node.relNode,
+										reln->smgr_rnode.backend);
+
+	v = _mdfd_getseg(reln, forknum, blocknum, false, EXTENSION_RETURN_NULL);
+
+	/* would need another segment */
+	if (v == NULL)
+		return 0;
+
+	seekpos = (off_t) BLCKSZ *(blocknum % ((BlockNumber) RELSEG_SIZE));
+
+	Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+
+	if (FileSeek(v->mdfd_vfd, seekpos, SEEK_SET) != seekpos)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not seek to block %u in file \"%s\": %m",
+						blocknum, FilePathName(v->mdfd_vfd))));
+
+	nbytes = FileRead(v->mdfd_vfd, buffer, BLCKSZ);
+
+	TRACE_POSTGRESQL_SMGR_MD_READ_DONE(forknum, blocknum,
+									   reln->smgr_rnode.node.spcNode,
+									   reln->smgr_rnode.node.dbNode,
+									   reln->smgr_rnode.node.relNode,
+									   reln->smgr_rnode.backend,
+									   nbytes,
+									   BLCKSZ);
+
+	if (nbytes < 0)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read block %u in file \"%s\": %m",
+						blocknum, FilePathName(v->mdfd_vfd))));
+
+	if (nbytes > 0 && nbytes < BLCKSZ)
+	{
+		ereport(LOG,
+				(errcode(ERRCODE_DATA_CORRUPTED),
+				 errmsg("could not read block %u in file \"%s\": read only %d of %d bytes",
+						blocknum, FilePathName(v->mdfd_vfd),
+						nbytes, BLCKSZ)));
+	}
+
+	return nbytes;
+}
+
 /*
  *	mdwrite() -- Write the supplied block at the appropriate location.
  *
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index 244b4ea..f0e9a7b 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -51,6 +51,8 @@ typedef struct f_smgr
 											  BlockNumber blocknum);
 	void		(*smgr_read) (SMgrRelation reln, ForkNumber forknum,
 										  BlockNumber blocknum, char *buffer);
+	int			(*smgr_tryread) (SMgrRelation reln, ForkNumber forknum,
+										  BlockNumber blocknum, char *buffer);
 	void		(*smgr_write) (SMgrRelation reln, ForkNumber forknum,
 						 BlockNumber blocknum, char *buffer, bool skipFsync);
 	BlockNumber (*smgr_nblocks) (SMgrRelation reln, ForkNumber forknum);
@@ -66,7 +68,7 @@ typedef struct f_smgr
 static const f_smgr smgrsw[] = {
 	/* magnetic disk */
 	{mdinit, NULL, mdclose, mdcreate, mdexists, mdunlink, mdextend,
-		mdprefetch, mdread, mdwrite, mdnblocks, mdtruncate, mdimmedsync,
+		mdprefetch, mdread, mdtryread, mdwrite, mdnblocks, mdtruncate, mdimmedsync,
 		mdpreckpt, mdsync, mdpostckpt
 	}
 };
@@ -626,6 +628,22 @@ smgrread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 	(*(smgrsw[reln->smgr_which].smgr_read)) (reln, forknum, blocknum, buffer);
 }
 
+
+/*
+ *	smgtryrread() -- read a particular block from a relation into the supplied
+ *				  buffer.
+ *
+ *		This routine is called from the buffer manager in order to
+ *		instantiate pages in the shared buffer cache.  All storage managers
+ *		return pages in the format that POSTGRES expects.
+ */
+int
+smgrtryread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+		 char *buffer)
+{
+	return (*(smgrsw[reln->smgr_which].smgr_tryread)) (reln, forknum, blocknum, buffer);
+}
+
 /*
  *	smgrwrite() -- Write the supplied buffer out.
  *
diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index 521ee1c..5f961af 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -40,6 +40,7 @@
 #define BM_CHECKPOINT_NEEDED	(1 << 7)		/* must write for checkpoint */
 #define BM_PERMANENT			(1 << 8)		/* permanent relation (not
 												 * unlogged) */
+#define BM_NEW					(1 << 9)		/* Not guaranteed to exist on disk */
 
 typedef bits16 BufFlags;
 
diff --git a/src/include/storage/bufmgr.h b/src/include/storage/bufmgr.h
index ec0a254..b52591f 100644
--- a/src/include/storage/bufmgr.h
+++ b/src/include/storage/bufmgr.h
@@ -153,6 +153,7 @@ extern Buffer ReadBufferExtended(Relation reln, ForkNumber forkNum,
 extern Buffer ReadBufferWithoutRelcache(RelFileNode rnode,
 						  ForkNumber forkNum, BlockNumber blockNum,
 						  ReadBufferMode mode, BufferAccessStrategy strategy);
+extern Buffer ExtendRelation(Relation reln, ForkNumber forkNum, BufferAccessStrategy strategy);
 extern void ReleaseBuffer(Buffer buffer);
 extern void UnlockReleaseBuffer(Buffer buffer);
 extern void MarkBufferDirty(Buffer buffer);
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index 69a624f..07a331c 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -94,6 +94,8 @@ extern void smgrprefetch(SMgrRelation reln, ForkNumber forknum,
 			 BlockNumber blocknum);
 extern void smgrread(SMgrRelation reln, ForkNumber forknum,
 		 BlockNumber blocknum, char *buffer);
+extern int smgrtryread(SMgrRelation reln, ForkNumber forknum,
+		 BlockNumber blocknum, char *buffer);
 extern void smgrwrite(SMgrRelation reln, ForkNumber forknum,
 		  BlockNumber blocknum, char *buffer, bool skipFsync);
 extern BlockNumber smgrnblocks(SMgrRelation reln, ForkNumber forknum);
@@ -114,12 +116,15 @@ extern void mdclose(SMgrRelation reln, ForkNumber forknum);
 extern void mdcreate(SMgrRelation reln, ForkNumber forknum, bool isRedo);
 extern bool mdexists(SMgrRelation reln, ForkNumber forknum);
 extern void mdunlink(RelFileNodeBackend rnode, ForkNumber forknum, bool isRedo);
-extern void mdextend(SMgrRelation reln, ForkNumber forknum,
+extern void mdextend(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, char *buffer, bool skipFsync);
+extern void mdappend(SMgrRelation reln, ForkNumber forknum,
 		 BlockNumber blocknum, char *buffer, bool skipFsync);
 extern void mdprefetch(SMgrRelation reln, ForkNumber forknum,
 		   BlockNumber blocknum);
 extern void mdread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 	   char *buffer);
+extern int mdtryread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+	   char *buffer);
 extern void mdwrite(SMgrRelation reln, ForkNumber forknum,
 		BlockNumber blocknum, char *buffer, bool skipFsync);
 extern BlockNumber mdnblocks(SMgrRelation reln, ForkNumber forknum);
-- 
2.3.0.149.gf3f4077.dirty

*** a/src/backend/access/brin/brin_pageops.c
--- b/src/backend/access/brin/brin_pageops.c
***************
*** 771,776 **** brin_getinsertbuffer(Relation irel, Buffer oldbuf, Size itemsz,
--- 771,781 ----
  			UnlockRelationForExtension(irel, ExclusiveLock);
  
  		page = BufferGetPage(buf);
+ 		if (PageIsNew(page))
+ 		{
+ 			MarkBufferDirty(buf);
+ 			PageInit(page, BufferGetPageSize(buf), 0);
+ 		}
  
  		/*
  		 * We have a new buffer to insert into.  Check that the new page has
*** a/src/backend/access/heap/hio.c
--- b/src/backend/access/heap/hio.c
***************
*** 393,398 **** RelationGetBufferForTuple(Relation relation, Size len,
--- 393,404 ----
  		 * we're done.
  		 */
  		page = BufferGetPage(buffer);
+ 		if (PageIsNew(page))
+ 		{
+ 			MarkBufferDirty(buffer);
+ 			PageInit(page, BufferGetPageSize(buffer), 0);
+ 		}
+ 
  		pageFreeSpace = PageGetHeapFreeSpace(page);
  		if (len + saveFreeSpace <= pageFreeSpace)
  		{
*** a/src/backend/storage/buffer/bufmgr.c
--- b/src/backend/storage/buffer/bufmgr.c
***************
*** 90,95 **** int			effective_io_concurrency = 0;
--- 90,96 ----
   * effective_io_concurrency parameter set.
   */
  int			target_prefetch_pages = 0;
+ int			extend_num_pages = 0;
  
  /* local state for StartBufferIO and related functions */
  static BufferDesc *InProgressBuf = NULL;
***************
*** 394,400 **** ForgetPrivateRefCountEntry(PrivateRefCountEntry *ref)
  static Buffer ReadBuffer_common(SMgrRelation reln, char relpersistence,
  				  ForkNumber forkNum, BlockNumber blockNum,
  				  ReadBufferMode mode, BufferAccessStrategy strategy,
! 				  bool *hit);
  static bool PinBuffer(BufferDesc *buf, BufferAccessStrategy strategy);
  static void PinBuffer_Locked(BufferDesc *buf);
  static void UnpinBuffer(BufferDesc *buf, bool fixOwner);
--- 395,401 ----
  static Buffer ReadBuffer_common(SMgrRelation reln, char relpersistence,
  				  ForkNumber forkNum, BlockNumber blockNum,
  				  ReadBufferMode mode, BufferAccessStrategy strategy,
! 				  bool *hit, Relation rel);
  static bool PinBuffer(BufferDesc *buf, BufferAccessStrategy strategy);
  static void PinBuffer_Locked(BufferDesc *buf);
  static void UnpinBuffer(BufferDesc *buf, bool fixOwner);
***************
*** 621,627 **** ReadBufferExtended(Relation reln, ForkNumber forkNum, BlockNumber blockNum,
  	 */
  	pgstat_count_buffer_read(reln);
  	buf = ReadBuffer_common(reln->rd_smgr, reln->rd_rel->relpersistence,
! 							forkNum, blockNum, mode, strategy, &hit);
  	if (hit)
  		pgstat_count_buffer_hit(reln);
  	return buf;
--- 622,628 ----
  	 */
  	pgstat_count_buffer_read(reln);
  	buf = ReadBuffer_common(reln->rd_smgr, reln->rd_rel->relpersistence,
! 							forkNum, blockNum, mode, strategy, &hit, reln);
  	if (hit)
  		pgstat_count_buffer_hit(reln);
  	return buf;
***************
*** 649,655 **** ReadBufferWithoutRelcache(RelFileNode rnode, ForkNumber forkNum,
  	Assert(InRecovery);
  
  	return ReadBuffer_common(smgr, RELPERSISTENCE_PERMANENT, forkNum, blockNum,
! 							 mode, strategy, &hit);
  }
  
  
--- 650,656 ----
  	Assert(InRecovery);
  
  	return ReadBuffer_common(smgr, RELPERSISTENCE_PERMANENT, forkNum, blockNum,
! 							 mode, strategy, &hit, NULL);
  }
  
  
***************
*** 661,667 **** ReadBufferWithoutRelcache(RelFileNode rnode, ForkNumber forkNum,
  static Buffer
  ReadBuffer_common(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
  				  BlockNumber blockNum, ReadBufferMode mode,
! 				  BufferAccessStrategy strategy, bool *hit)
  {
  	BufferDesc *bufHdr;
  	Block		bufBlock;
--- 662,668 ----
  static Buffer
  ReadBuffer_common(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
  				  BlockNumber blockNum, ReadBufferMode mode,
! 				  BufferAccessStrategy strategy, bool *hit, Relation rel)
  {
  	BufferDesc *bufHdr;
  	Block		bufBlock;
***************
*** 685,691 **** ReadBuffer_common(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
--- 686,695 ----
  
  	/* Substitute proper block number if caller asked for P_NEW */
  	if (isExtend)
+ 	{
  		blockNum = smgrnblocks(smgr, forkNum);
+ 		//blockNum += extend_num_pages;
+ 	}
  
  	if (isLocalBuf)
  	{
***************
*** 814,823 **** ReadBuffer_common(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
--- 818,836 ----
  
  	if (isExtend)
  	{
+ 		int blkCount = 0;
+ 
  		/* new buffers are zero-filled */
  		MemSet((char *) bufBlock, 0, BLCKSZ);
  		/* don't set checksum for all-zero page */
  		smgrextend(smgr, forkNum, blockNum, (char *) bufBlock, false);
+ 
+ 		while (blkCount < extend_num_pages)
+ 		{
+ 			blkCount++;
+ 			smgrextend(smgr, forkNum, blockNum+blkCount, (char *) bufBlock, false);
+ 			RecordPageWithFreeSpace(rel, blockNum+blkCount, 8126);
+ 		}
  	}
  	else
  	{
*** a/src/backend/utils/misc/guc.c
--- b/src/backend/utils/misc/guc.c
***************
*** 2683,2688 **** static struct config_int ConfigureNamesInt[] =
--- 2683,2698 ----
  		NULL, NULL, NULL
  	},
  
+ 	{
+ 		{"extend_num_pages", PGC_SUSET, RESOURCES_ASYNCHRONOUS,
+ 			gettext_noop("Sets the Number of pages to extended at one time."),
+ 			NULL
+ 		},
+ 		&extend_num_pages,
+ 		0, 0, 100,
+ 		NULL, NULL, NULL
+ 	},
+ 
  	/* End-of-list marker */
  	{
  		{NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL
*** a/src/backend/utils/misc/postgresql.conf.sample
--- b/src/backend/utils/misc/postgresql.conf.sample
***************
*** 139,144 ****
--- 139,146 ----
  
  #temp_file_limit = -1			# limits per-session temp file space
  					# in kB, or -1 for no limit
+ #extend_num_pages = 0			# number of extra pages allocate during extend
+ 					# min 0 max 100 pages
  
  # - Kernel Resource Usage -
  
*** a/src/include/storage/bufmgr.h
--- b/src/include/storage/bufmgr.h
***************
*** 60,65 **** extern PGDLLIMPORT char *BufferBlocks;
--- 60,66 ----
  
  /* in guc.c */
  extern int	effective_io_concurrency;
+ extern int	extend_num_pages;
  
  /* in localbuf.c */
  extern PGDLLIMPORT int NLocBuffer;

*** a/src/backend/access/heap/hio.c
--- b/src/backend/access/heap/hio.c
***************
*** 24,29 ****
--- 24,30 ----
  #include "storage/lmgr.h"
  #include "storage/smgr.h"
  
+ int extend_num_pages = 0;
  
  /*
   * RelationPutHeapTuple - place tuple at specified page
***************
*** 238,243 **** RelationGetBufferForTuple(Relation relation, Size len,
--- 239,245 ----
  	BlockNumber targetBlock,
  				otherBlock;
  	bool		needLock;
+ 	int			totalBlocks;
  
  	len = MAXALIGN(len);		/* be conservative */
  
***************
*** 449,467 **** RelationGetBufferForTuple(Relation relation, Size len,
  	 * it worth keeping an accurate file length in shared memory someplace,
  	 * rather than relying on the kernel to do it for us?
  	 */
! 	buffer = ReadBufferBI(relation, P_NEW, bistate);
! 
! 	/*
! 	 * We can be certain that locking the otherBuffer first is OK, since it
! 	 * must have a lower page number.
! 	 */
! 	if (otherBuffer != InvalidBuffer)
! 		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
! 
! 	/*
! 	 * Now acquire lock on the new page.
! 	 */
! 	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
  
  	/*
  	 * Release the file-extension lock; it's now OK for someone else to extend
--- 451,491 ----
  	 * it worth keeping an accurate file length in shared memory someplace,
  	 * rather than relying on the kernel to do it for us?
  	 */
! 
! 	totalBlocks = extend_num_pages;
! 
! 	do {
! 
! 
! 		buffer = ReadBufferBI(relation, P_NEW, bistate);
! 
! 		/*
! 		 * We can be certain that locking the otherBuffer first is OK, since it
! 		 * must have a lower page number.
! 		 */
! 		if ((otherBuffer != InvalidBuffer) && !totalBlocks)
! 			LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
! 
! 		/*
! 		 * Now acquire lock on the new page.
! 		 */
! 		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
! 
! 		if (totalBlocks)
! 		{
! 			Page page;
! 			Size freespace;
! 
! 			page = BufferGetPage(buffer);
! 			PageInit(page, BufferGetPageSize(buf), 0);
! 
! 			freespace = PageGetHeapFreeSpace(page);
! 			MarkBufferDirty(buffer);
! 			UnlockReleaseBuffer(buffer);
! 			RecordPageWithFreeSpace(relation, BufferGetBlockNumber(buffer), freespace);
! 		}
! 
! 	}while (totalBlocks--);
  
  	/*
  	 * Release the file-extension lock; it's now OK for someone else to extend
*** a/src/backend/utils/misc/guc.c
--- b/src/backend/utils/misc/guc.c
***************
*** 31,36 ****
--- 31,37 ----
  #include "access/transam.h"
  #include "access/twophase.h"
  #include "access/xact.h"
+ #include "access/hio.h"
  #include "catalog/namespace.h"
  #include "commands/async.h"
  #include "commands/prepare.h"
***************
*** 2683,2688 **** static struct config_int ConfigureNamesInt[] =
--- 2684,2699 ----
  		NULL, NULL, NULL
  	},
  
+ 	{
+ 		{"extend_num_pages", PGC_USERSET, UNGROUPED,
+ 			gettext_noop("Sets the Number of pages to extended at one time."),
+ 			NULL
+ 		},
+ 		&extend_num_pages,
+ 		0, 0, 100,
+ 		NULL, NULL, NULL
+ 	},
+ 
  	/* End-of-list marker */
  	{
  		{NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL
*** a/src/backend/utils/misc/postgresql.conf.sample
--- b/src/backend/utils/misc/postgresql.conf.sample
***************
*** 139,144 ****
--- 139,146 ----
  
  #temp_file_limit = -1			# limits per-session temp file space
  					# in kB, or -1 for no limit
+ #extend_num_pages = 0			# number of extra pages allocate during extend
+ 					# min 0 max 100 pages
  
  # - Kernel Resource Usage -
  
*** a/src/include/access/hio.h
--- b/src/include/access/hio.h
***************
*** 19,25 ****
  #include "utils/relcache.h"
  #include "storage/buf.h"
  
! 
  /*
   * state for bulk inserts --- private to heapam.c and hio.c
   *
--- 19,25 ----
  #include "utils/relcache.h"
  #include "storage/buf.h"
  
! extern int extend_num_pages;
  /*
   * state for bulk inserts --- private to heapam.c and hio.c
   *

diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index 86b9ae1..76f9a21 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -268,6 +268,16 @@ static relopt_int intRelOpts[] =
 #endif
 	},
 
+	{
+		{
+			"extend_by_blocks",
+			"Number of blocks to be added to relation in every extend call",
+			RELOPT_KIND_HEAP,
+			AccessExclusiveLock
+		},
+		1, 1, 100
+	},
+
 	/* list terminator */
 	{{NULL}}
 };
@@ -1291,7 +1301,9 @@ default_reloptions(Datum reloptions, bool validate, relopt_kind kind)
 		{"autovacuum_analyze_scale_factor", RELOPT_TYPE_REAL,
 		offsetof(StdRdOptions, autovacuum) +offsetof(AutoVacOpts, analyze_scale_factor)},
 		{"user_catalog_table", RELOPT_TYPE_BOOL,
-		offsetof(StdRdOptions, user_catalog_table)}
+		offsetof(StdRdOptions, user_catalog_table)},
+		{"extend_by_blocks", RELOPT_TYPE_INT,
+		offsetof(StdRdOptions, extend_by_blocks)}
 	};
 
 	options = parseRelOptions(reloptions, validate, kind, &numoptions);
diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..ec430fc 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -238,6 +238,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	BlockNumber targetBlock,
 				otherBlock;
 	bool		needLock;
+	int			extraBlocks;
 
 	len = MAXALIGN(len);		/* be conservative */
 
@@ -443,25 +444,49 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	if (needLock)
 		LockRelationForExtension(relation, ExclusiveLock);
 
+	if (use_fsm)
+		extraBlocks = RelationGetExtendBlocks(relation) -1;
+	else
+		extraBlocks = 0;
 	/*
 	 * XXX This does an lseek - rather expensive - but at the moment it is the
 	 * only way to accurately determine how many blocks are in a relation.  Is
 	 * it worth keeping an accurate file length in shared memory someplace,
 	 * rather than relying on the kernel to do it for us?
 	 */
-	buffer = ReadBufferBI(relation, P_NEW, bistate);
 
-	/*
-	 * We can be certain that locking the otherBuffer first is OK, since it
-	 * must have a lower page number.
-	 */
-	if (otherBuffer != InvalidBuffer)
-		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+	do {
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
 
-	/*
-	 * Now acquire lock on the new page.
-	 */
-	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+		/*
+		 * We can be certain that locking the otherBuffer first is OK, since it
+		 * must have a lower page number.
+		 */
+		if ((otherBuffer != InvalidBuffer) && !extraBlocks)
+			LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+
+		/*
+		 * Now acquire lock on the new page.
+		 */
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+		if (extraBlocks)
+		{
+			Page page;
+			Size freespace;
+			BlockNumber blockNum;
+
+			page = BufferGetPage(buffer);
+			PageInit(page, BufferGetPageSize(buffer), 0);
+
+			freespace = PageGetHeapFreeSpace(page);
+			MarkBufferDirty(buffer);
+			blockNum = BufferGetBlockNumber(buffer);
+			UnlockReleaseBuffer(buffer);
+			RecordPageWithFreeSpace(relation, blockNum, freespace);
+		}
+
+	}while (extraBlocks--);
 
 	/*
 	 * Release the file-extension lock; it's now OK for someone else to extend
diff --git a/src/include/access/hio.h b/src/include/access/hio.h
index a174b34..40c3941 100644
--- a/src/include/access/hio.h
+++ b/src/include/access/hio.h
@@ -19,7 +19,6 @@
 #include "utils/relcache.h"
 #include "storage/buf.h"
 
-
 /*
  * state for bulk inserts --- private to heapam.c and hio.c
  *
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index f2bebf2..26f6b8e 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -203,6 +203,7 @@ typedef struct StdRdOptions
 	AutoVacOpts autovacuum;		/* autovacuum-related options */
 	bool		user_catalog_table;		/* use as an additional catalog
 										 * relation */
+	int			extend_by_blocks;
 } StdRdOptions;
 
 #define HEAP_MIN_FILLFACTOR			10
@@ -239,6 +240,13 @@ typedef struct StdRdOptions
 	((relation)->rd_options ?				\
 	 ((StdRdOptions *) (relation)->rd_options)->user_catalog_table : false)
 
+/*
+ * RelationGetExtendBlocks
+ *		Returns the relation's number of block to be extended one time.
+ */
+#define RelationGetExtendBlocks(relation) \
+	((relation)->rd_options ? \
+	 ((StdRdOptions *) (relation)->rd_options)->extend_by_blocks : 1)
 
 /*
  * ViewOptions

diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index 86b9ae1..78e81dd 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -268,6 +268,16 @@ static relopt_int intRelOpts[] =
 #endif
 	},
 
+	{
+		{
+			"extend_by_blocks",
+			"Number of blocks to be added to relation in every extend call",
+			RELOPT_KIND_HEAP,
+			AccessExclusiveLock
+		},
+		1, 1, 10000
+	},
+
 	/* list terminator */
 	{{NULL}}
 };
@@ -1291,7 +1301,9 @@ default_reloptions(Datum reloptions, bool validate, relopt_kind kind)
 		{"autovacuum_analyze_scale_factor", RELOPT_TYPE_REAL,
 		offsetof(StdRdOptions, autovacuum) +offsetof(AutoVacOpts, analyze_scale_factor)},
 		{"user_catalog_table", RELOPT_TYPE_BOOL,
-		offsetof(StdRdOptions, user_catalog_table)}
+		offsetof(StdRdOptions, user_catalog_table)},
+		{"extend_by_blocks", RELOPT_TYPE_INT,
+		offsetof(StdRdOptions, extend_by_blocks)}
 	};
 
 	options = parseRelOptions(reloptions, validate, kind, &numoptions);
diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..eb3ce17 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -238,6 +238,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	BlockNumber targetBlock,
 				otherBlock;
 	bool		needLock;
+	int			extraBlocks;
 
 	len = MAXALIGN(len);		/* be conservative */
 
@@ -443,25 +444,50 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	if (needLock)
 		LockRelationForExtension(relation, ExclusiveLock);
 
+	if (use_fsm)
+		extraBlocks = RelationGetExtendBlocks(relation) - 1;
+	else
+		extraBlocks = 0;
 	/*
 	 * XXX This does an lseek - rather expensive - but at the moment it is the
 	 * only way to accurately determine how many blocks are in a relation.  Is
 	 * it worth keeping an accurate file length in shared memory someplace,
 	 * rather than relying on the kernel to do it for us?
 	 */
-	buffer = ReadBufferBI(relation, P_NEW, bistate);
 
-	/*
-	 * We can be certain that locking the otherBuffer first is OK, since it
-	 * must have a lower page number.
-	 */
-	if (otherBuffer != InvalidBuffer)
-		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+	do
+	{
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
 
-	/*
-	 * Now acquire lock on the new page.
-	 */
-	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+		/*
+		 * We can be certain that locking the otherBuffer first is OK, since
+		 * it must have a lower page number.
+		 */
+		if ((otherBuffer != InvalidBuffer) && !extraBlocks)
+			LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+
+		/*
+		 * Now acquire lock on the new page.
+		 */
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+		if (extraBlocks)
+		{
+			Page		page;
+			Size		freespace;
+			BlockNumber blockNum;
+
+			page = BufferGetPage(buffer);
+			PageInit(page, BufferGetPageSize(buffer), 0);
+
+			freespace = PageGetHeapFreeSpace(page);
+			MarkBufferDirty(buffer);
+			blockNum = BufferGetBlockNumber(buffer);
+			UnlockReleaseBuffer(buffer);
+			RecordPageWithFreeSpace(relation, blockNum, freespace);
+		}
+
+	} while (extraBlocks--);
 
 	/*
 	 * Release the file-extension lock; it's now OK for someone else to extend
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index f2bebf2..26f6b8e 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -203,6 +203,7 @@ typedef struct StdRdOptions
 	AutoVacOpts autovacuum;		/* autovacuum-related options */
 	bool		user_catalog_table;		/* use as an additional catalog
 										 * relation */
+	int			extend_by_blocks;
 } StdRdOptions;
 
 #define HEAP_MIN_FILLFACTOR			10
@@ -239,6 +240,13 @@ typedef struct StdRdOptions
 	((relation)->rd_options ?				\
 	 ((StdRdOptions *) (relation)->rd_options)->user_catalog_table : false)
 
+/*
+ * RelationGetExtendBlocks
+ *		Returns the relation's number of block to be extended one time.
+ */
+#define RelationGetExtendBlocks(relation) \
+	((relation)->rd_options ? \
+	 ((StdRdOptions *) (relation)->rd_options)->extend_by_blocks : 1)
 
 /*
  * ViewOptions

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..adb82ba 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -23,6 +23,7 @@
 #include "storage/freespace.h"
 #include "storage/lmgr.h"
 #include "storage/smgr.h"
+#include "storage/proc.h"
 
 
 /*
@@ -168,6 +169,160 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 	}
 }
 
+
+static BlockNumber
+GroupExtendRelation(PGPROC *proc, Relation relation, BulkInsertState bistate)
+{
+	volatile PROC_HDR *procglobal = ProcGlobal;
+	uint32		nextidx;
+	uint32		wakeidx;
+	int			extraWaits = -1;
+	BlockNumber targetBlock;
+	int count = 0;
+
+	/* Add ourselves to the list of processes needing a group extend. */
+	proc->groupExtendMember = true;
+
+	while (true)
+	{
+		nextidx = pg_atomic_read_u32(&procglobal->extendGroupFirst);
+		pg_atomic_write_u32(&proc->extendGroupNext, nextidx);
+
+		if (pg_atomic_compare_exchange_u32(&procglobal->extendGroupFirst,
+										   &nextidx,
+										   (uint32) proc->pgprocno))
+			break;
+	}
+
+	/*
+	 * If the list was not empty, the leader will clear our XID.  It is
+	 * impossible to have followers without a leader because the first process
+	 * that has added itself to the list will always have nextidx as
+	 * INVALID_PGPROCNO.
+	 */
+	if (nextidx != INVALID_PGPROCNO)
+	{
+		/* Sleep until the leader clears our XID. */
+		for (;;)
+		{
+			/* acts as a read barrier */
+			PGSemaphoreLock(&proc->sem);
+			if (!proc->groupExtendMember)
+				break;
+			extraWaits++;
+		}
+
+		Assert(pg_atomic_read_u32(&proc->extendGroupNext) == INVALID_PGPROCNO);
+
+		/* Fix semaphore count for any absorbed wake ups */
+		while (extraWaits-- > 0)
+			PGSemaphoreUnlock(&proc->sem);
+
+		targetBlock = proc->blockNum;
+
+		proc->blockNum = InvalidBlockNumber;
+
+		return targetBlock;
+	}
+
+	/* We are the leader.  Acquire the lock on behalf of everyone. */
+	LockRelationForExtension(relation, ExclusiveLock);
+
+	/*
+	 * Now that we've got the lock, clear the list of processes waiting for
+	 * group extending
+	 */
+	while (true)
+	{
+		nextidx = pg_atomic_read_u32(&procglobal->extendGroupFirst);
+		if (pg_atomic_compare_exchange_u32(&procglobal->extendGroupFirst,
+										   &nextidx,
+										   PG_INT32_MAX))
+			break;
+	}
+
+	/* Remember head of list so we can perform wakeups after dropping lock. */
+	wakeidx = nextidx;
+
+
+	/* Walk the list and clear all XIDs. */
+	while (nextidx != INVALID_PGPROCNO)
+	{
+		PGPROC	   *proc = &ProcGlobal->allProcs[nextidx];
+		Buffer buffer;
+
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+		proc->blockNum = BufferGetBlockNumber(buffer);
+
+		ReleaseBuffer(buffer);
+
+		/* Move to next proc in list. */
+		nextidx = pg_atomic_read_u32(&proc->extendGroupNext);
+
+		count ++;
+	}
+
+	count = count*10;
+
+	do
+	{
+		Buffer buffer;
+		Page		page;
+		Size		freespace;
+		BlockNumber blockNum;
+
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+		/*
+		 * Now acquire lock on the new page.
+		 */
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+		page = BufferGetPage(buffer);
+		PageInit(page, BufferGetPageSize(buffer), 0);
+
+		freespace = PageGetHeapFreeSpace(page);
+		MarkBufferDirty(buffer);
+		blockNum = BufferGetBlockNumber(buffer);
+		UnlockReleaseBuffer(buffer);
+
+		RecordPageWithFreeSpace(relation, blockNum, freespace);
+	} while (count--);
+
+	/* We're done with the lock now. */
+	UnlockRelationForExtension(relation, ExclusiveLock);
+
+	/*
+	 * Now that we've released the lock, go back and wake everybody up.  We
+	 * don't do this under the lock so as to keep lock hold times to a
+	 * minimum.  The system calls we need to perform to wake other processes
+	 * up are probably much slower than the simple memory writes we did while
+	 * holding the lock.
+	 */
+	while (wakeidx != INVALID_PGPROCNO)
+	{
+		PGPROC	   *proc = &ProcGlobal->allProcs[wakeidx];
+
+		wakeidx = pg_atomic_read_u32(&proc->extendGroupNext);
+		pg_atomic_write_u32(&proc->extendGroupNext, INVALID_PGPROCNO);
+
+		/* ensure all previous writes are visible before follower continues. */
+		pg_write_barrier();
+
+		proc->groupExtendMember = false;
+
+		if (proc != MyProc)
+			PGSemaphoreUnlock(&proc->sem);
+		else
+		{
+			targetBlock = proc->blockNum;
+			proc->blockNum = InvalidBlockNumber;
+		}
+	}
+
+	return targetBlock;
+}
+
 /*
  * RelationGetBufferForTuple
  *
@@ -238,6 +393,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	BlockNumber targetBlock,
 				otherBlock;
 	bool		needLock;
+	int			extraBlocks;
 
 	len = MAXALIGN(len);		/* be conservative */
 
@@ -308,6 +464,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		}
 	}
 
+loop:
 	while (targetBlock != InvalidBlockNumber)
 	{
 		/*
@@ -441,27 +598,95 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	needLock = !RELATION_IS_LOCAL(relation);
 
 	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
+	{
+		if (TryLockRelationForExtension(relation, ExclusiveLock))
+		{
+			buffer = ReadBufferBI(relation, P_NEW, bistate);
+			/*
+			 * We can be certain that locking the otherBuffer first is OK, since
+			 * it must have a lower page number.
+			 */
+			if ((otherBuffer != InvalidBuffer) && !extraBlocks)
+				LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+
+			/*
+			 * Now acquire lock on the new page.
+			 */
+			LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+			UnlockRelationForExtension(relation, ExclusiveLock);
+		}
+		else
+		{
+			targetBlock = GroupExtendRelation(MyProc, relation, bistate);
+			goto loop;
+		}
+	}
+	else
+	{
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+		/*
+		 * We can be certain that locking the otherBuffer first is OK, since
+		 * it must have a lower page number.
+		 */
+		if ((otherBuffer != InvalidBuffer) && !extraBlocks)
+			LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
 
+		/*
+		 * Now acquire lock on the new page.
+		 */
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+	}
+
+#if 0
+	if (LOCKACQUIRE_NOT_AVAIL)
+
+	if (use_fsm)
+		extraBlocks = RelationGetExtendBlocks(relation) - 1;
+	else
+		extraBlocks = 0;
 	/*
 	 * XXX This does an lseek - rather expensive - but at the moment it is the
 	 * only way to accurately determine how many blocks are in a relation.  Is
 	 * it worth keeping an accurate file length in shared memory someplace,
 	 * rather than relying on the kernel to do it for us?
 	 */
-	buffer = ReadBufferBI(relation, P_NEW, bistate);
 
-	/*
-	 * We can be certain that locking the otherBuffer first is OK, since it
-	 * must have a lower page number.
-	 */
-	if (otherBuffer != InvalidBuffer)
-		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+	do
+	{
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+		/*
+		 * We can be certain that locking the otherBuffer first is OK, since
+		 * it must have a lower page number.
+		 */
+		if ((otherBuffer != InvalidBuffer) && !extraBlocks)
+			LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+
+		/*
+		 * Now acquire lock on the new page.
+		 */
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+		if (extraBlocks)
+		{
+			Page		page;
+			Size		freespace;
+			BlockNumber blockNum;
+
+			page = BufferGetPage(buffer);
+			PageInit(page, BufferGetPageSize(buffer), 0);
+
+			freespace = PageGetHeapFreeSpace(page);
+			MarkBufferDirty(buffer);
+			blockNum = BufferGetBlockNumber(buffer);
+			UnlockReleaseBuffer(buffer);
+			RecordPageWithFreeSpace(relation, blockNum, freespace);
+		}
+
+	} while (extraBlocks--);
 
-	/*
-	 * Now acquire lock on the new page.
-	 */
-	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
 
 	/*
 	 * Release the file-extension lock; it's now OK for someone else to extend
@@ -471,7 +696,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	 */
 	if (needLock)
 		UnlockRelationForExtension(relation, ExclusiveLock);
-
+#endif
 	/*
 	 * We need to initialize the empty new page.  Double-check that it really
 	 * is empty (this should never happen, but if it does we don't want to
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 9d16afb..08e5e07 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -340,6 +340,21 @@ LockRelationForExtension(Relation relation, LOCKMODE lockmode)
 	(void) LockAcquire(&tag, lockmode, false, false);
 }
 
+LockAcquireResult
+TryLockRelationForExtension(Relation relation, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockAcquire(&tag, lockmode, false, true);
+}
+
+
+
+
 /*
  *		UnlockRelationForExtension
  */
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index 6453b88..a823eda 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -182,6 +182,7 @@ InitProcGlobal(void)
 	ProcGlobal->walwriterLatch = NULL;
 	ProcGlobal->checkpointerLatch = NULL;
 	pg_atomic_init_u32(&ProcGlobal->procArrayGroupFirst, INVALID_PGPROCNO);
+	pg_atomic_init_u32(&ProcGlobal->extendGroupFirst, INVALID_PGPROCNO);
 
 	/*
 	 * Create and initialize all the PGPROC structures we'll need.  There are
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index dbcdd3f..5a0e60b 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -152,6 +152,11 @@ struct PGPROC
 	 */
 	TransactionId	procArrayGroupMemberXid;
 
+	bool groupExtendMember;
+	pg_atomic_uint32	extendGroupNext;
+	uint32	blockNum;
+
+
 	/* Per-backend LWLock.  Protects fields below (but not group fields). */
 	LWLock		backendLock;
 
@@ -223,6 +228,8 @@ typedef struct PROC_HDR
 	PGPROC	   *bgworkerFreeProcs;
 	/* First pgproc waiting for group XID clear */
 	pg_atomic_uint32 procArrayGroupFirst;
+	pg_atomic_uint32 extendGroupFirst;
+
 	/* WALWriter process's latch */
 	Latch	   *walwriterLatch;
 	/* Checkpointer process's latch */

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..832a80d 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -167,6 +167,32 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 			break;
 	}
 }
+static Buffer
+RelationAddOneBlock(Relation relation, Buffer otherBuffer, BulkInsertState bistate)
+{
+	Buffer buffer;
+	/*
+	 * XXX This does an lseek - rather expensive - but at the moment it is the
+	 * only way to accurately determine how many blocks are in a relation.  Is
+	 * it worth keeping an accurate file length in shared memory someplace,
+	 * rather than relying on the kernel to do it for us?
+	 */
+	buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+	/*
+	 * We can be certain that locking the otherBuffer first is OK, since
+	 * it must have a lower page number.
+	 */
+	if (otherBuffer != InvalidBuffer)
+		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+
+	/*
+	 * Now acquire lock on the new page.
+	 */
+	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+	return buffer;
+}
 
 /*
  * RelationGetBufferForTuple
@@ -236,7 +262,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	Size		pageFreeSpace,
 				saveFreeSpace;
 	BlockNumber targetBlock,
-				otherBlock;
+				otherBlock,
+				lastValidBlock;
 	bool		needLock;
 
 	len = MAXALIGN(len);		/* be conservative */
@@ -308,6 +335,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		}
 	}
 
+loop:
 	while (targetBlock != InvalidBlockNumber)
 	{
 		/*
@@ -388,6 +416,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 								 otherBlock, targetBlock, vmbuffer_other,
 								 vmbuffer);
 
+		lastValidBlock = targetBlock;
+
 		/*
 		 * Now we can check to see if there's enough free space here. If so,
 		 * we're done.
@@ -441,36 +471,76 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	needLock = !RELATION_IS_LOCAL(relation);
 
 	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
-
-	/*
-	 * XXX This does an lseek - rather expensive - but at the moment it is the
-	 * only way to accurately determine how many blocks are in a relation.  Is
-	 * it worth keeping an accurate file length in shared memory someplace,
-	 * rather than relying on the kernel to do it for us?
-	 */
-	buffer = ReadBufferBI(relation, P_NEW, bistate);
-
-	/*
-	 * We can be certain that locking the otherBuffer first is OK, since it
-	 * must have a lower page number.
-	 */
-	if (otherBuffer != InvalidBuffer)
-		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
-
-	/*
-	 * Now acquire lock on the new page.
-	 */
-	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
-
-	/*
-	 * Release the file-extension lock; it's now OK for someone else to extend
-	 * the relation some more.  Note that we cannot release this lock before
-	 * we have buffer lock on the new page, or we risk a race condition
-	 * against vacuumlazy.c --- see comments therein.
-	 */
-	if (needLock)
-		UnlockRelationForExtension(relation, ExclusiveLock);
+	{
+		if (TryLockRelationForExtension(relation, ExclusiveLock))
+		{
+			buffer = RelationAddOneBlock(relation, otherBuffer, bistate);
+			UnlockRelationForExtension(relation, ExclusiveLock);
+		}
+		else
+		{
+			LockRelationForExtension(relation, ExclusiveLock);
+
+			if (use_fsm)
+			{
+				Page		page;
+				Size		freespace;
+				BlockNumber blockNum;
+				int			extraBlocks = 0;
+				int 		lockWaiters = 0;
+				Buffer		buf;
+
+				/*
+				 * Update FSM as to condition of this page, and ask for another page
+				 * to try.
+				 */
+				targetBlock = RecordAndGetPageWithFreeSpace(relation,
+															lastValidBlock,
+															pageFreeSpace,
+															len + saveFreeSpace);
+
+				/* Other waiter has extended the block for us*/
+				if (targetBlock != InvalidBlockNumber)
+				{
+					UnlockRelationForExtension(relation, ExclusiveLock);
+					goto loop;
+				}
+
+				buffer = RelationAddOneBlock(relation, otherBuffer, bistate);
+
+				lockWaiters = RelationExtensionLockWaiter(relation);
+
+				extraBlocks = lockWaiters * 20;
+
+				while (extraBlocks--)
+				{
+					/*
+					 * XXX This does an lseek - rather expensive - but at the moment it is the
+					 * only way to accurately determine how many blocks are in a relation.  Is
+					 * it worth keeping an accurate file length in shared memory someplace,
+					 * rather than relying on the kernel to do it for us?
+					 */
+					buf = ReadBufferBI(relation, P_NEW, bistate);
+					LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
+
+					page = BufferGetPage(buf);
+					PageInit(page, BufferGetPageSize(buf), 0);
+					freespace = PageGetHeapFreeSpace(page);
+					MarkBufferDirty(buf);
+					blockNum = BufferGetBlockNumber(buf);
+					UnlockReleaseBuffer(buf);
+
+					RecordPageWithFreeSpace(relation, blockNum, freespace);
+				}
+			}
+			else
+				buffer = RelationAddOneBlock(relation, otherBuffer, bistate);
+
+			UnlockRelationForExtension(relation, ExclusiveLock);
+		}
+	}
+	else
+		buffer = RelationAddOneBlock(relation, otherBuffer, bistate);
 
 	/*
 	 * We need to initialize the empty new page.  Double-check that it really
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 9d16afb..a56b203 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -340,6 +340,30 @@ LockRelationForExtension(Relation relation, LOCKMODE lockmode)
 	(void) LockAcquire(&tag, lockmode, false, false);
 }
 
+LockAcquireResult
+TryLockRelationForExtension(Relation relation, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockAcquire(&tag, lockmode, false, true);
+}
+
+int
+RelationExtensionLockWaiter(Relation relation)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockWaiterCount(&tag);
+}
+
 /*
  *		UnlockRelationForExtension
  */
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index a458c68..8f49192 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -4380,3 +4380,35 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	LockRelease(&tag, ShareLock, false);
 	return true;
 }
+
+int
+LockWaiterCount(const LOCKTAG *locktag)
+{
+	LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
+	LOCALLOCKTAG localtag;
+	LOCALLOCK  *locallock;
+	LOCK	   *lock;
+	bool		found;
+	uint32		hashcode;
+	int			waiters = 0;
+
+	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
+		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
+
+
+	/*
+	 * Find a LOCALLOCK entry for this lock and lockmode
+	 */
+	MemSet(&localtag, 0, sizeof(localtag));		/* must clear padding */
+	localtag.lock = *locktag;
+	localtag.mode = ExclusiveLock;
+
+	locallock = (LOCALLOCK *) hash_search(LockMethodLocalHash,
+										  (void *) &localtag,
+										  HASH_FIND, &found);
+
+	if (found)
+		waiters = locallock->lock->nRequested;
+
+	return waiters;
+}
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index e9d41bf..a492bb1 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -101,4 +101,7 @@ extern void UnlockSharedObjectForSession(Oid classid, Oid objid, uint16 objsubid
 /* Describe a locktag for error messages */
 extern void DescribeLockTag(StringInfo buf, const LOCKTAG *tag);
 
+extern LockAcquireResult TryLockRelationForExtension(Relation relation, LOCKMODE lockmode);
+extern int RelationExtensionLockWaiter(Relation relation);
+
 #endif   /* LMGR_H */
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index 788d50a..3fd74fb 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -572,6 +572,8 @@ extern void RememberSimpleDeadLock(PGPROC *proc1,
 					   PGPROC *proc2);
 extern void InitDeadLockChecking(void);
 
+extern int LockWaiterCount(const LOCKTAG *locktag);
+
 #ifdef LOCK_DEBUG
 extern void DumpLocks(PGPROC *proc);
 extern void DumpAllLocks(void);

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..eb4ee0c 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -169,6 +169,86 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 }
 
 /*
+ * RelationAddOneBlock
+ *
+ * Extend relation by one block and lock the buffer
+ */
+static Buffer
+RelationAddOneBlock(Relation relation, Buffer otherBuffer, BulkInsertState bistate)
+{
+	Buffer buffer;
+	/*
+	 * XXX This does an lseek - rather expensive - but at the moment it is the
+	 * only way to accurately determine how many blocks are in a relation.  Is
+	 * it worth keeping an accurate file length in shared memory someplace,
+	 * rather than relying on the kernel to do it for us?
+	 */
+	buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+	/*
+	 * We can be certain that locking the otherBuffer first is OK, since
+	 * it must have a lower page number.
+	 */
+	if (otherBuffer != InvalidBuffer)
+		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+
+	/*
+	 * Now acquire lock on the new page.
+	 */
+	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+	return buffer;
+}
+/*
+ * RelationAddExtraBlocks
+ *
+ * Extend extra blocks for the relations to avoid the future contention
+ * on the relation extension lock.
+ */
+
+static void
+RelationAddExtraBlocks(Relation relation, BulkInsertState bistate)
+{
+	Page		page;
+	Size		freespace;
+	BlockNumber blockNum;
+	int			extraBlocks = 0;
+	int 		lockWaiters = 0;
+	Buffer		buffer;
+
+	/*
+	 * For calculating number of extra blocks to extend, find the level
+	 * of contention on this lock, by getting the requester of this lock
+	 * and add extra blocks in multiple of waiters.
+	 */
+	lockWaiters = RelationExtensionLockWaiter(relation);
+
+	extraBlocks = lockWaiters * 20;
+
+	while (extraBlocks--)
+	{
+		/*
+		 * Here we are adding extra blocks to the relation after
+		 * adding each block update the information in FSM so that
+		 * other backend running parallel can find the block.
+		 */
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+		page = BufferGetPage(buffer);
+		PageInit(page, BufferGetPageSize(buffer), 0);
+		freespace = PageGetHeapFreeSpace(page);
+		MarkBufferDirty(buffer);
+		blockNum = BufferGetBlockNumber(buffer);
+
+		UnlockReleaseBuffer(buffer);
+
+		RecordPageWithFreeSpace(relation, blockNum, freespace);
+	}
+}
+
+/*
  * RelationGetBufferForTuple
  *
  *	Returns pinned and exclusive-locked buffer of a page in given relation
@@ -233,10 +313,11 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	bool		use_fsm = !(options & HEAP_INSERT_SKIP_FSM);
 	Buffer		buffer = InvalidBuffer;
 	Page		page;
-	Size		pageFreeSpace,
-				saveFreeSpace;
+	Size		pageFreeSpace = 0,
+				saveFreeSpace = 0;
 	BlockNumber targetBlock,
-				otherBlock;
+				otherBlock,
+				lastValidBlock = InvalidBlockNumber;
 	bool		needLock;
 
 	len = MAXALIGN(len);		/* be conservative */
@@ -308,6 +389,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		}
 	}
 
+loop:
 	while (targetBlock != InvalidBlockNumber)
 	{
 		/*
@@ -388,6 +470,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 								 otherBlock, targetBlock, vmbuffer_other,
 								 vmbuffer);
 
+		lastValidBlock = targetBlock;
+
 		/*
 		 * Now we can check to see if there's enough free space here. If so,
 		 * we're done.
@@ -441,37 +525,53 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	needLock = !RELATION_IS_LOCAL(relation);
 
 	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
-
-	/*
-	 * XXX This does an lseek - rather expensive - but at the moment it is the
-	 * only way to accurately determine how many blocks are in a relation.  Is
-	 * it worth keeping an accurate file length in shared memory someplace,
-	 * rather than relying on the kernel to do it for us?
-	 */
-	buffer = ReadBufferBI(relation, P_NEW, bistate);
-
-	/*
-	 * We can be certain that locking the otherBuffer first is OK, since it
-	 * must have a lower page number.
-	 */
-	if (otherBuffer != InvalidBuffer)
-		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+	{
+		/*
+		 * First try to get the lock in no-wait mode, if succeed extend one
+		 * block, else get the lock in normal mode and after we get the lock
+		 * extend some extra blocks, extra blocks will be added to satisfy
+		 * request of other waiters and avoid future contention. Here instead
+		 * of directly taking lock we try no-wait mode, this is to handle the
+		 * case, when there is no contention - it should not find the lock
+		 * waiter and execute extra instructions.
+		 */
+		if (LOCKACQUIRE_OK
+				!= RelationExtensionLockConditional(relation, ExclusiveLock))
+		{
+			LockRelationForExtension(relation, ExclusiveLock);
+			
+			if (use_fsm)
+			{
+				if (lastValidBlock != InvalidBlockNumber)
+				{
+					targetBlock = RecordAndGetPageWithFreeSpace(relation,
+														lastValidBlock,
+														pageFreeSpace,
+														len + saveFreeSpace);
+				}	
+
+				/* Other waiter has extended the block for us*/
+				if (targetBlock != InvalidBlockNumber)
+				{
+					UnlockRelationForExtension(relation, ExclusiveLock);
+					goto loop;
+				}
+				
+				RelationAddExtraBlocks(relation, bistate);
+			}
+		}
+	}
 
-	/*
-	 * Now acquire lock on the new page.
+	/* For all case we need to add at least one block to satisfy our
+	 * own request.
 	 */
-	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+	buffer = RelationAddOneBlock(relation, otherBuffer, bistate);
 
-	/*
-	 * Release the file-extension lock; it's now OK for someone else to extend
-	 * the relation some more.  Note that we cannot release this lock before
-	 * we have buffer lock on the new page, or we risk a race condition
-	 * against vacuumlazy.c --- see comments therein.
-	 */
 	if (needLock)
 		UnlockRelationForExtension(relation, ExclusiveLock);
 
+
+
 	/*
 	 * We need to initialize the empty new page.  Double-check that it really
 	 * is empty (this should never happen, but if it does we don't want to
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 9d16afb..5d9454d 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -341,6 +341,40 @@ LockRelationForExtension(Relation relation, LOCKMODE lockmode)
 }
 
 /*
+ *		RelationExtensionLockConditional
+ *
+ * Same as LockRelationForExtension except it will not wait on the lock.
+ */
+LockAcquireResult
+RelationExtensionLockConditional(Relation relation, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockAcquire(&tag, lockmode, false, true);
+}
+
+/*
+ *		RelationExtensionLockWaiter
+ *
+ * Count the lock requester for the RelationExtension lock.
+ */
+int
+RelationExtensionLockWaiter(Relation relation)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockWaiterCount(&tag);
+}
+
+/*
  *		UnlockRelationForExtension
  */
 void
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index a458c68..cf723ae 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -4380,3 +4380,47 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	LockRelease(&tag, ShareLock, false);
 	return true;
 }
+
+/*
+ * LockWaiterCount
+ *
+ * Find the number of lock requester on this locktag
+ */
+int
+LockWaiterCount(const LOCKTAG *locktag)
+{
+	LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
+	LOCALLOCKTAG localtag;
+	LOCALLOCK  *locallock;
+	bool		found;
+	uint32		hashcode;
+	LWLock		*partitionLock;
+	int			waiters = 0;
+
+	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
+		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
+
+	/*
+	 * Find a LOCALLOCK entry for this lock and lockmode
+	 */
+	MemSet(&localtag, 0, sizeof(localtag));		/* must clear padding */
+	localtag.lock = *locktag;
+	localtag.mode = ExclusiveLock;
+
+	locallock = (LOCALLOCK *) hash_search(LockMethodLocalHash,
+										(void *) &localtag,
+										HASH_FIND, &found);
+
+
+	hashcode = locallock->hashcode;
+
+	partitionLock = LockHashPartitionLock(hashcode);
+	LWLockAcquire(partitionLock, LW_EXCLUSIVE);
+
+	if (found)
+		waiters = locallock->lock->nRequested;
+
+	LWLockRelease(partitionLock);
+
+	return waiters;
+}
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index e9d41bf..ea8e19e 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -101,4 +101,7 @@ extern void UnlockSharedObjectForSession(Oid classid, Oid objid, uint16 objsubid
 /* Describe a locktag for error messages */
 extern void DescribeLockTag(StringInfo buf, const LOCKTAG *tag);
 
+extern LockAcquireResult RelationExtensionLockConditional(Relation relation, LOCKMODE lockmode);
+extern int RelationExtensionLockWaiter(Relation relation);
+
 #endif   /* LMGR_H */
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index 788d50a..3fd74fb 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -572,6 +572,8 @@ extern void RememberSimpleDeadLock(PGPROC *proc1,
 					   PGPROC *proc2);
 extern void InitDeadLockChecking(void);
 
+extern int LockWaiterCount(const LOCKTAG *locktag);
+
 #ifdef LOCK_DEBUG
 extern void DumpLocks(PGPROC *proc);
 extern void DumpAllLocks(void);

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..b73535c 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -169,6 +169,87 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 }
 
 /*
+ * RelationAddOneBlock
+ *
+ * Extend relation by one block and lock the buffer
+ */
+static Buffer
+RelationAddOneBlock(Relation relation, Buffer otherBuffer, BulkInsertState bistate)
+{
+	Buffer buffer;
+	/*
+	 * XXX This does an lseek - rather expensive - but at the moment it is the
+	 * only way to accurately determine how many blocks are in a relation.  Is
+	 * it worth keeping an accurate file length in shared memory someplace,
+	 * rather than relying on the kernel to do it for us?
+	 */
+	buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+	/*
+	 * We can be certain that locking the otherBuffer first is OK, since
+	 * it must have a lower page number.
+	 */
+	if (otherBuffer != InvalidBuffer)
+		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+
+	/*
+	 * Now acquire lock on the new page.
+	 */
+	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+	return buffer;
+}
+
+/*
+ * RelationAddExtraBlocks
+ *
+ * Extend extra blocks for the relations to avoid the future contention
+ * on the relation extension lock.
+ */
+
+static void
+RelationAddExtraBlocks(Relation relation, BulkInsertState bistate)
+{
+	Page		page;
+	Size		freespace;
+	BlockNumber blockNum;
+	int			extraBlocks = 0;
+	int 		lockWaiters = 0;
+	Buffer		buffer;
+
+	/*
+	 * For calculating number of extra blocks to extend, find the level
+	 * of contention on this lock, by getting the requester of this lock
+	 * and add extra blocks in multiple of waiters.
+	 */
+	lockWaiters = RelationExtensionLockWaiter(relation);
+
+	extraBlocks = lockWaiters * 20;
+
+	while (extraBlocks--)
+	{
+		/*
+		 * Here we are adding extra blocks to the relation after
+		 * adding each block update the information in FSM so that
+		 * other backend running parallel can find the block.
+		 */
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+		page = BufferGetPage(buffer);
+		PageInit(page, BufferGetPageSize(buffer), 0);
+		freespace = PageGetHeapFreeSpace(page);
+		MarkBufferDirty(buffer);
+		blockNum = BufferGetBlockNumber(buffer);
+
+		UnlockReleaseBuffer(buffer);
+
+		RecordPageWithFreeSpace(relation, blockNum, freespace);
+	}
+}
+
+/*
  * RelationGetBufferForTuple
  *
  *	Returns pinned and exclusive-locked buffer of a page in given relation
@@ -233,10 +314,11 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	bool		use_fsm = !(options & HEAP_INSERT_SKIP_FSM);
 	Buffer		buffer = InvalidBuffer;
 	Page		page;
-	Size		pageFreeSpace,
-				saveFreeSpace;
+	Size		pageFreeSpace = 0,
+				saveFreeSpace = 0;
 	BlockNumber targetBlock,
-				otherBlock;
+				otherBlock,
+				lastValidBlock = InvalidBlockNumber;
 	bool		needLock;
 
 	len = MAXALIGN(len);		/* be conservative */
@@ -308,6 +390,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		}
 	}
 
+loop:
 	while (targetBlock != InvalidBlockNumber)
 	{
 		/*
@@ -388,6 +471,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 								 otherBlock, targetBlock, vmbuffer_other,
 								 vmbuffer);
 
+		lastValidBlock = targetBlock;
+
 		/*
 		 * Now we can check to see if there's enough free space here. If so,
 		 * we're done.
@@ -441,27 +526,47 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	needLock = !RELATION_IS_LOCAL(relation);
 
 	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
-
-	/*
-	 * XXX This does an lseek - rather expensive - but at the moment it is the
-	 * only way to accurately determine how many blocks are in a relation.  Is
-	 * it worth keeping an accurate file length in shared memory someplace,
-	 * rather than relying on the kernel to do it for us?
-	 */
-	buffer = ReadBufferBI(relation, P_NEW, bistate);
-
-	/*
-	 * We can be certain that locking the otherBuffer first is OK, since it
-	 * must have a lower page number.
-	 */
-	if (otherBuffer != InvalidBuffer)
-		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+	{
+		/*
+		 * First try to get the lock in no-wait mode, if succeed extend one
+		 * block, else get the lock in normal mode and after we get the lock
+		 * extend some extra blocks, extra blocks will be added to satisfy
+		 * request of other waiters and avoid future contention. Here instead
+		 * of directly taking lock we try no-wait mode, this is to handle the
+		 * case, when there is no contention - it should not find the lock
+		 * waiter and execute extra instructions.
+		 */
+		if (LOCKACQUIRE_OK
+				!= RelationExtensionLockConditional(relation, ExclusiveLock))
+		{
+			LockRelationForExtension(relation, ExclusiveLock);
+
+			if (use_fsm)
+			{
+				if (lastValidBlock != InvalidBlockNumber)
+				{
+					targetBlock = RecordAndGetPageWithFreeSpace(relation,
+														lastValidBlock,
+														pageFreeSpace,
+														len + saveFreeSpace);
+				}
+
+				/* Other waiter has extended the block for us*/
+				if (targetBlock != InvalidBlockNumber)
+				{
+					UnlockRelationForExtension(relation, ExclusiveLock);
+					goto loop;
+				}
+
+				RelationAddExtraBlocks(relation, bistate);
+			}
+		}
+	}
 
-	/*
-	 * Now acquire lock on the new page.
+	/* For all case we need to add at least one block to satisfy our
+	 * own request.
 	 */
-	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+	buffer = RelationAddOneBlock(relation, otherBuffer, bistate);
 
 	/*
 	 * Release the file-extension lock; it's now OK for someone else to extend
@@ -472,6 +577,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	if (needLock)
 		UnlockRelationForExtension(relation, ExclusiveLock);
 
+
+
 	/*
 	 * We need to initialize the empty new page.  Double-check that it really
 	 * is empty (this should never happen, but if it does we don't want to
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 9d16afb..5d9454d 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -341,6 +341,40 @@ LockRelationForExtension(Relation relation, LOCKMODE lockmode)
 }
 
 /*
+ *		RelationExtensionLockConditional
+ *
+ * Same as LockRelationForExtension except it will not wait on the lock.
+ */
+LockAcquireResult
+RelationExtensionLockConditional(Relation relation, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockAcquire(&tag, lockmode, false, true);
+}
+
+/*
+ *		RelationExtensionLockWaiter
+ *
+ * Count the lock requester for the RelationExtension lock.
+ */
+int
+RelationExtensionLockWaiter(Relation relation)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockWaiterCount(&tag);
+}
+
+/*
  *		UnlockRelationForExtension
  */
 void
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index a458c68..cf723ae 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -4380,3 +4380,47 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	LockRelease(&tag, ShareLock, false);
 	return true;
 }
+
+/*
+ * LockWaiterCount
+ *
+ * Find the number of lock requester on this locktag
+ */
+int
+LockWaiterCount(const LOCKTAG *locktag)
+{
+	LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
+	LOCALLOCKTAG localtag;
+	LOCALLOCK  *locallock;
+	bool		found;
+	uint32		hashcode;
+	LWLock		*partitionLock;
+	int			waiters = 0;
+
+	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
+		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
+
+	/*
+	 * Find a LOCALLOCK entry for this lock and lockmode
+	 */
+	MemSet(&localtag, 0, sizeof(localtag));		/* must clear padding */
+	localtag.lock = *locktag;
+	localtag.mode = ExclusiveLock;
+
+	locallock = (LOCALLOCK *) hash_search(LockMethodLocalHash,
+										(void *) &localtag,
+										HASH_FIND, &found);
+
+
+	hashcode = locallock->hashcode;
+
+	partitionLock = LockHashPartitionLock(hashcode);
+	LWLockAcquire(partitionLock, LW_EXCLUSIVE);
+
+	if (found)
+		waiters = locallock->lock->nRequested;
+
+	LWLockRelease(partitionLock);
+
+	return waiters;
+}
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index e9d41bf..ea8e19e 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -101,4 +101,7 @@ extern void UnlockSharedObjectForSession(Oid classid, Oid objid, uint16 objsubid
 /* Describe a locktag for error messages */
 extern void DescribeLockTag(StringInfo buf, const LOCKTAG *tag);
 
+extern LockAcquireResult RelationExtensionLockConditional(Relation relation, LOCKMODE lockmode);
+extern int RelationExtensionLockWaiter(Relation relation);
+
 #endif   /* LMGR_H */
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index 788d50a..3fd74fb 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -572,6 +572,8 @@ extern void RememberSimpleDeadLock(PGPROC *proc1,
 					   PGPROC *proc2);
 extern void InitDeadLockChecking(void);
 
+extern int LockWaiterCount(const LOCKTAG *locktag);
+
 #ifdef LOCK_DEBUG
 extern void DumpLocks(PGPROC *proc);
 extern void DumpAllLocks(void);

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..fcd42b9 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -169,6 +169,94 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 }
 
 /*
+ * RelationAddOneBlock
+ *
+ * Extend relation by one block and lock the buffer
+ */
+static Buffer
+RelationAddOneBlock(Relation relation, Buffer otherBuffer, BulkInsertState bistate)
+{
+	Buffer buffer;
+	/*
+	 * XXX This does an lseek - rather expensive - but at the moment it is the
+	 * only way to accurately determine how many blocks are in a relation.  Is
+	 * it worth keeping an accurate file length in shared memory someplace,
+	 * rather than relying on the kernel to do it for us?
+	 */
+	buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+	/*
+	 * We can be certain that locking the otherBuffer first is OK, since
+	 * it must have a lower page number.
+	 */
+	if (otherBuffer != InvalidBuffer)
+		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+
+	/*
+	 * Now acquire lock on the new page.
+	 */
+	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+	return buffer;
+}
+
+/*
+ * RelationAddExtraBlocks
+ *
+ * Extend extra blocks for the relations to avoid the future contention
+ * on the relation extension lock.
+ */
+
+static void
+RelationAddExtraBlocks(Relation relation, BulkInsertState bistate)
+{
+	Page		page;
+	Size		freespace;
+	BlockNumber blockNum;
+	int			extraBlocks = 0;
+	int 		lockWaiters = 0;
+	Buffer		buffer;
+
+	/*
+	 * For calculating number of extra blocks to extend, find the level
+	 * of contention on this lock, by getting the requester of this lock
+	 * and add extra blocks in multiple of waiters.
+	 */
+	lockWaiters = RelationExtensionLockWaiter(relation);
+
+	extraBlocks = lockWaiters * 20;
+
+	/* To avoid the cases when there are huge number of lock waiters, and
+	 * extend file size by big amount at a time, put some limit on the
+	 * max number of pages to be extended at a time.
+	 */
+	if (extraBlocks > 512)
+		extraBlocks = 512;
+
+	while (extraBlocks--)
+	{
+		/*
+		 * Here we are adding extra blocks to the relation after
+		 * adding each block update the information in FSM so that
+		 * other backend running parallel can find the block.
+		 */
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+		page = BufferGetPage(buffer);
+		PageInit(page, BufferGetPageSize(buffer), 0);
+		freespace = PageGetHeapFreeSpace(page);
+		MarkBufferDirty(buffer);
+		blockNum = BufferGetBlockNumber(buffer);
+
+		UnlockReleaseBuffer(buffer);
+
+		RecordPageWithFreeSpace(relation, blockNum, freespace);
+	}
+}
+
+/*
  * RelationGetBufferForTuple
  *
  *	Returns pinned and exclusive-locked buffer of a page in given relation
@@ -233,10 +321,11 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	bool		use_fsm = !(options & HEAP_INSERT_SKIP_FSM);
 	Buffer		buffer = InvalidBuffer;
 	Page		page;
-	Size		pageFreeSpace,
-				saveFreeSpace;
+	Size		pageFreeSpace = 0,
+				saveFreeSpace = 0;
 	BlockNumber targetBlock,
-				otherBlock;
+				otherBlock,
+				lastValidBlock = InvalidBlockNumber;
 	bool		needLock;
 
 	len = MAXALIGN(len);		/* be conservative */
@@ -308,6 +397,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		}
 	}
 
+loop:
 	while (targetBlock != InvalidBlockNumber)
 	{
 		/*
@@ -388,6 +478,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 								 otherBlock, targetBlock, vmbuffer_other,
 								 vmbuffer);
 
+		lastValidBlock = targetBlock;
+
 		/*
 		 * Now we can check to see if there's enough free space here. If so,
 		 * we're done.
@@ -441,27 +533,47 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	needLock = !RELATION_IS_LOCAL(relation);
 
 	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
-
-	/*
-	 * XXX This does an lseek - rather expensive - but at the moment it is the
-	 * only way to accurately determine how many blocks are in a relation.  Is
-	 * it worth keeping an accurate file length in shared memory someplace,
-	 * rather than relying on the kernel to do it for us?
-	 */
-	buffer = ReadBufferBI(relation, P_NEW, bistate);
-
-	/*
-	 * We can be certain that locking the otherBuffer first is OK, since it
-	 * must have a lower page number.
-	 */
-	if (otherBuffer != InvalidBuffer)
-		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+	{
+		/*
+		 * First try to get the lock in no-wait mode, if succeed extend one
+		 * block, else get the lock in normal mode and after we get the lock
+		 * extend some extra blocks, extra blocks will be added to satisfy
+		 * request of other waiters and avoid future contention. Here instead
+		 * of directly taking lock we try no-wait mode, this is to handle the
+		 * case, when there is no contention - it should not find the lock
+		 * waiter and execute extra instructions.
+		 */
+		if (LOCKACQUIRE_OK
+				!= RelationExtensionLockConditional(relation, ExclusiveLock))
+		{
+			LockRelationForExtension(relation, ExclusiveLock);
+
+			if (use_fsm)
+			{
+				if (lastValidBlock != InvalidBlockNumber)
+				{
+					targetBlock = RecordAndGetPageWithFreeSpace(relation,
+														lastValidBlock,
+														pageFreeSpace,
+														len + saveFreeSpace);
+				}
+
+				/* Other waiter has extended the block for us*/
+				if (targetBlock != InvalidBlockNumber)
+				{
+					UnlockRelationForExtension(relation, ExclusiveLock);
+					goto loop;
+				}
+
+				RelationAddExtraBlocks(relation, bistate);
+			}
+		}
+	}
 
-	/*
-	 * Now acquire lock on the new page.
+	/* For all case we need to add at least one block to satisfy our
+	 * own request.
 	 */
-	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+	buffer = RelationAddOneBlock(relation, otherBuffer, bistate);
 
 	/*
 	 * Release the file-extension lock; it's now OK for someone else to extend
@@ -472,6 +584,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	if (needLock)
 		UnlockRelationForExtension(relation, ExclusiveLock);
 
+
+
 	/*
 	 * We need to initialize the empty new page.  Double-check that it really
 	 * is empty (this should never happen, but if it does we don't want to
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 9d16afb..5d9454d 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -341,6 +341,40 @@ LockRelationForExtension(Relation relation, LOCKMODE lockmode)
 }
 
 /*
+ *		RelationExtensionLockConditional
+ *
+ * Same as LockRelationForExtension except it will not wait on the lock.
+ */
+LockAcquireResult
+RelationExtensionLockConditional(Relation relation, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockAcquire(&tag, lockmode, false, true);
+}
+
+/*
+ *		RelationExtensionLockWaiter
+ *
+ * Count the lock requester for the RelationExtension lock.
+ */
+int
+RelationExtensionLockWaiter(Relation relation)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockWaiterCount(&tag);
+}
+
+/*
  *		UnlockRelationForExtension
  */
 void
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index a458c68..cf723ae 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -4380,3 +4380,47 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	LockRelease(&tag, ShareLock, false);
 	return true;
 }
+
+/*
+ * LockWaiterCount
+ *
+ * Find the number of lock requester on this locktag
+ */
+int
+LockWaiterCount(const LOCKTAG *locktag)
+{
+	LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
+	LOCALLOCKTAG localtag;
+	LOCALLOCK  *locallock;
+	bool		found;
+	uint32		hashcode;
+	LWLock		*partitionLock;
+	int			waiters = 0;
+
+	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
+		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
+
+	/*
+	 * Find a LOCALLOCK entry for this lock and lockmode
+	 */
+	MemSet(&localtag, 0, sizeof(localtag));		/* must clear padding */
+	localtag.lock = *locktag;
+	localtag.mode = ExclusiveLock;
+
+	locallock = (LOCALLOCK *) hash_search(LockMethodLocalHash,
+										(void *) &localtag,
+										HASH_FIND, &found);
+
+
+	hashcode = locallock->hashcode;
+
+	partitionLock = LockHashPartitionLock(hashcode);
+	LWLockAcquire(partitionLock, LW_EXCLUSIVE);
+
+	if (found)
+		waiters = locallock->lock->nRequested;
+
+	LWLockRelease(partitionLock);
+
+	return waiters;
+}
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index e9d41bf..ea8e19e 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -101,4 +101,7 @@ extern void UnlockSharedObjectForSession(Oid classid, Oid objid, uint16 objsubid
 /* Describe a locktag for error messages */
 extern void DescribeLockTag(StringInfo buf, const LOCKTAG *tag);
 
+extern LockAcquireResult RelationExtensionLockConditional(Relation relation, LOCKMODE lockmode);
+extern int RelationExtensionLockWaiter(Relation relation);
+
 #endif   /* LMGR_H */
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index 788d50a..3fd74fb 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -572,6 +572,8 @@ extern void RememberSimpleDeadLock(PGPROC *proc1,
 					   PGPROC *proc2);
 extern void InitDeadLockChecking(void);
 
+extern int LockWaiterCount(const LOCKTAG *locktag);
+
 #ifdef LOCK_DEBUG
 extern void DumpLocks(PGPROC *proc);
 extern void DumpAllLocks(void);

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..a9e18dc 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -169,6 +169,91 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 }
 
 /*
+ * RelationAddOneBlock
+ *
+ * Extend relation by one block and lock the buffer
+ */
+static Buffer
+RelationAddOneBlock(Relation relation, Buffer otherBuffer, BulkInsertState bistate)
+{
+	Buffer buffer;
+	/*
+	 * XXX This does an lseek - rather expensive - but at the moment it is the
+	 * only way to accurately determine how many blocks are in a relation.  Is
+	 * it worth keeping an accurate file length in shared memory someplace,
+	 * rather than relying on the kernel to do it for us?
+	 */
+	buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+	/*
+	 * We can be certain that locking the otherBuffer first is OK, since
+	 * it must have a lower page number.
+	 */
+	if (otherBuffer != InvalidBuffer)
+		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+
+	/*
+	 * Now acquire lock on the new page.
+	 */
+	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+	return buffer;
+}
+
+/*
+ * RelationAddExtraBlocks
+ *
+ * Extend extra blocks for the relations to avoid the future contention
+ * on the relation extension lock.
+ */
+
+static void
+RelationAddExtraBlocks(Relation relation, BulkInsertState bistate)
+{
+	Page		page;
+	Size		freespace;
+	BlockNumber blockNum;
+	int			extraBlocks = 0;
+	int 		lockWaiters = 0;
+	Buffer		buffer;
+
+	/*
+	 * For calculating number of extra blocks to extend, find the level
+	 * of contention on this lock, by getting the requester of this lock
+	 * and add extra blocks in multiple of waiters.
+	 */
+	lockWaiters = RelationExtensionLockWaiter(relation);
+
+	/* To avoid the cases when there are huge number of lock waiters, and
+	 * extend file size by big amount at a time, put some limit on the
+	 * max number of pages to be extended at a time.
+	 */
+	extraBlocks = Min(512, lockWaiters * 20);
+
+	while (extraBlocks--)
+	{
+		/*
+		 * Here we are adding extra blocks to the relation after
+		 * adding each block update the information in FSM so that
+		 * other backend running parallel can find the block.
+		 */
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+		page = BufferGetPage(buffer);
+		PageInit(page, BufferGetPageSize(buffer), 0);
+		freespace = PageGetHeapFreeSpace(page);
+		MarkBufferDirty(buffer);
+		blockNum = BufferGetBlockNumber(buffer);
+
+		UnlockReleaseBuffer(buffer);
+
+		RecordPageWithFreeSpace(relation, blockNum, freespace);
+	}
+}
+
+/*
  * RelationGetBufferForTuple
  *
  *	Returns pinned and exclusive-locked buffer of a page in given relation
@@ -233,10 +318,11 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	bool		use_fsm = !(options & HEAP_INSERT_SKIP_FSM);
 	Buffer		buffer = InvalidBuffer;
 	Page		page;
-	Size		pageFreeSpace,
-				saveFreeSpace;
+	Size		pageFreeSpace = 0,
+				saveFreeSpace = 0;
 	BlockNumber targetBlock,
-				otherBlock;
+				otherBlock,
+				lastValidBlock = InvalidBlockNumber;
 	bool		needLock;
 
 	len = MAXALIGN(len);		/* be conservative */
@@ -308,6 +394,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		}
 	}
 
+loop:
 	while (targetBlock != InvalidBlockNumber)
 	{
 		/*
@@ -388,6 +475,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 								 otherBlock, targetBlock, vmbuffer_other,
 								 vmbuffer);
 
+		lastValidBlock = targetBlock;
+
 		/*
 		 * Now we can check to see if there's enough free space here. If so,
 		 * we're done.
@@ -441,27 +530,47 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	needLock = !RELATION_IS_LOCAL(relation);
 
 	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
-
-	/*
-	 * XXX This does an lseek - rather expensive - but at the moment it is the
-	 * only way to accurately determine how many blocks are in a relation.  Is
-	 * it worth keeping an accurate file length in shared memory someplace,
-	 * rather than relying on the kernel to do it for us?
-	 */
-	buffer = ReadBufferBI(relation, P_NEW, bistate);
-
-	/*
-	 * We can be certain that locking the otherBuffer first is OK, since it
-	 * must have a lower page number.
-	 */
-	if (otherBuffer != InvalidBuffer)
-		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+	{
+		/*
+		 * First try to get the lock in no-wait mode, if succeed extend one
+		 * block, else get the lock in normal mode and after we get the lock
+		 * extend some extra blocks, extra blocks will be added to satisfy
+		 * request of other waiters and avoid future contention. Here instead
+		 * of directly taking lock we try no-wait mode, this is to handle the
+		 * case, when there is no contention - it should not find the lock
+		 * waiter and execute extra instructions.
+		 */
+		if (LOCKACQUIRE_OK
+				!= RelationExtensionLockConditional(relation, ExclusiveLock))
+		{
+			LockRelationForExtension(relation, ExclusiveLock);
+
+			if (use_fsm)
+			{
+				if (lastValidBlock != InvalidBlockNumber)
+				{
+					targetBlock = RecordAndGetPageWithFreeSpace(relation,
+														lastValidBlock,
+														pageFreeSpace,
+														len + saveFreeSpace);
+				}
+
+				/* Other waiter has extended the block for us*/
+				if (targetBlock != InvalidBlockNumber)
+				{
+					UnlockRelationForExtension(relation, ExclusiveLock);
+					goto loop;
+				}
+
+				RelationAddExtraBlocks(relation, bistate);
+			}
+		}
+	}
 
-	/*
-	 * Now acquire lock on the new page.
+	/* For all case we need to add at least one block to satisfy our
+	 * own request.
 	 */
-	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+	buffer = RelationAddOneBlock(relation, otherBuffer, bistate);
 
 	/*
 	 * Release the file-extension lock; it's now OK for someone else to extend
@@ -472,6 +581,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	if (needLock)
 		UnlockRelationForExtension(relation, ExclusiveLock);
 
+
+
 	/*
 	 * We need to initialize the empty new page.  Double-check that it really
 	 * is empty (this should never happen, but if it does we don't want to
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 9d16afb..5d9454d 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -341,6 +341,40 @@ LockRelationForExtension(Relation relation, LOCKMODE lockmode)
 }
 
 /*
+ *		RelationExtensionLockConditional
+ *
+ * Same as LockRelationForExtension except it will not wait on the lock.
+ */
+LockAcquireResult
+RelationExtensionLockConditional(Relation relation, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockAcquire(&tag, lockmode, false, true);
+}
+
+/*
+ *		RelationExtensionLockWaiter
+ *
+ * Count the lock requester for the RelationExtension lock.
+ */
+int
+RelationExtensionLockWaiter(Relation relation)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockWaiterCount(&tag);
+}
+
+/*
  *		UnlockRelationForExtension
  */
 void
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index a458c68..cf723ae 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -4380,3 +4380,47 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	LockRelease(&tag, ShareLock, false);
 	return true;
 }
+
+/*
+ * LockWaiterCount
+ *
+ * Find the number of lock requester on this locktag
+ */
+int
+LockWaiterCount(const LOCKTAG *locktag)
+{
+	LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
+	LOCALLOCKTAG localtag;
+	LOCALLOCK  *locallock;
+	bool		found;
+	uint32		hashcode;
+	LWLock		*partitionLock;
+	int			waiters = 0;
+
+	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
+		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
+
+	/*
+	 * Find a LOCALLOCK entry for this lock and lockmode
+	 */
+	MemSet(&localtag, 0, sizeof(localtag));		/* must clear padding */
+	localtag.lock = *locktag;
+	localtag.mode = ExclusiveLock;
+
+	locallock = (LOCALLOCK *) hash_search(LockMethodLocalHash,
+										(void *) &localtag,
+										HASH_FIND, &found);
+
+
+	hashcode = locallock->hashcode;
+
+	partitionLock = LockHashPartitionLock(hashcode);
+	LWLockAcquire(partitionLock, LW_EXCLUSIVE);
+
+	if (found)
+		waiters = locallock->lock->nRequested;
+
+	LWLockRelease(partitionLock);
+
+	return waiters;
+}
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index e9d41bf..ea8e19e 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -101,4 +101,7 @@ extern void UnlockSharedObjectForSession(Oid classid, Oid objid, uint16 objsubid
 /* Describe a locktag for error messages */
 extern void DescribeLockTag(StringInfo buf, const LOCKTAG *tag);
 
+extern LockAcquireResult RelationExtensionLockConditional(Relation relation, LOCKMODE lockmode);
+extern int RelationExtensionLockWaiter(Relation relation);
+
 #endif   /* LMGR_H */
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index 788d50a..3fd74fb 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -572,6 +572,8 @@ extern void RememberSimpleDeadLock(PGPROC *proc1,
 					   PGPROC *proc2);
 extern void InitDeadLockChecking(void);
 
+extern int LockWaiterCount(const LOCKTAG *locktag);
+
 #ifdef LOCK_DEBUG
 extern void DumpLocks(PGPROC *proc);
 extern void DumpAllLocks(void);

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..b1e764c 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -169,6 +169,102 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 }
 
 /*
+ * RelationAddOneBlock
+ *
+ * Extend relation by one block and lock the buffer, and caller must take care
+ * of unlocking the buffer.
+ */
+static Buffer
+RelationAddOneBlock(Relation relation, Buffer otherBuffer, BulkInsertState bistate)
+{
+	Buffer buffer;
+	/*
+	 * XXX This does an lseek - rather expensive - but at the moment it is the
+	 * only way to accurately determine how many blocks are in a relation.  Is
+	 * it worth keeping an accurate file length in shared memory someplace,
+	 * rather than relying on the kernel to do it for us?
+	 */
+	buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+	/*
+	 * We can be certain that locking the otherBuffer first is OK, since
+	 * it must have a lower page number.
+	 */
+	if (otherBuffer != InvalidBuffer)
+		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+
+	/*
+	 * Now acquire lock on the new page.
+	 */
+	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+	return buffer;
+}
+
+/*
+ * RelationAddExtraBlocks
+ *
+ * Extend extra blocks for the relations to avoid the future contention
+ * on the relation extension lock.
+ */
+
+static void
+RelationAddExtraBlocks(Relation relation, BulkInsertState bistate)
+{
+	Page		page;
+	Size		freespace;
+	BlockNumber blockNum;
+	int			extraBlocks = 0;
+	int 		lockWaiters = 0;
+	Buffer		buffer;
+
+	/*
+	 * For calculating number of extra blocks to extend, find the level
+	 * of contention on this lock, by getting the requester of this lock
+	 * and add extra blocks in multiple of waiters.
+	 */
+	lockWaiters = RelationExtensionLockWaiterCount(relation);
+
+	/*
+	 * To avoid the cases when there are huge number of lock waiters, and
+	 * extend file size by big amount at a time, put some limit on the
+	 * max number of pages to be extended at a time.
+	 * At a time we are extending (20 * LockWaiters), We reached to this
+	 * number by testing with different multipliers i.e 10,20,50.. and found
+	 * that 10 is good for the small tuple size but not sufficient when
+	 * tuple size is very huge (1K), and with 20 we get good scalability
+	 * with small tuple as well as with big tuple and increasing multiplier
+	 * beyond 20 doen't seems to be giving much benefit, So no point in
+	 * adding many extra block without seeing significant improvement.
+	 * Also we are limiting the extension at 512 blocks (4MB), Even though
+	 * there can be some benefit its better to have some limit.
+	 */
+	extraBlocks = Min(512, lockWaiters * 20);
+
+	while (extraBlocks--)
+	{
+		/*
+		 * Here we are adding extra blocks to the relation after
+		 * adding each block update the information in FSM so that
+		 * other backend running parallel can find the block.
+		 */
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+		page = BufferGetPage(buffer);
+		PageInit(page, BufferGetPageSize(buffer), 0);
+		freespace = PageGetHeapFreeSpace(page);
+		MarkBufferDirty(buffer);
+		blockNum = BufferGetBlockNumber(buffer);
+
+		UnlockReleaseBuffer(buffer);
+
+		RecordPageWithFreeSpace(relation, blockNum, freespace);
+	}
+}
+
+/*
  * RelationGetBufferForTuple
  *
  *	Returns pinned and exclusive-locked buffer of a page in given relation
@@ -233,10 +329,11 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	bool		use_fsm = !(options & HEAP_INSERT_SKIP_FSM);
 	Buffer		buffer = InvalidBuffer;
 	Page		page;
-	Size		pageFreeSpace,
-				saveFreeSpace;
+	Size		pageFreeSpace = 0,
+				saveFreeSpace = 0;
 	BlockNumber targetBlock,
-				otherBlock;
+				otherBlock,
+				lastValidBlock = InvalidBlockNumber;
 	bool		needLock;
 
 	len = MAXALIGN(len);		/* be conservative */
@@ -308,6 +405,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		}
 	}
 
+loop:
 	while (targetBlock != InvalidBlockNumber)
 	{
 		/*
@@ -388,6 +486,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 								 otherBlock, targetBlock, vmbuffer_other,
 								 vmbuffer);
 
+		lastValidBlock = targetBlock;
+
 		/*
 		 * Now we can check to see if there's enough free space here. If so,
 		 * we're done.
@@ -441,27 +541,58 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	needLock = !RELATION_IS_LOCAL(relation);
 
 	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
-
-	/*
-	 * XXX This does an lseek - rather expensive - but at the moment it is the
-	 * only way to accurately determine how many blocks are in a relation.  Is
-	 * it worth keeping an accurate file length in shared memory someplace,
-	 * rather than relying on the kernel to do it for us?
-	 */
-	buffer = ReadBufferBI(relation, P_NEW, bistate);
-
-	/*
-	 * We can be certain that locking the otherBuffer first is OK, since it
-	 * must have a lower page number.
-	 */
-	if (otherBuffer != InvalidBuffer)
-		LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+	{
+		/*
+		 * First try to get the lock in no-wait mode, if succeed extend one
+		 * block, else get the lock in normal mode and after we get the lock
+		 * extend some extra blocks, extra blocks will be added to satisfy
+		 * request of other waiters and avoid future contention. Here instead
+		 * of directly taking lock we try no-wait mode, this is to handle the
+		 * case, when there is no contention - it should not find the lock
+		 * waiter and execute extra instructions.
+		 */
+		if (RelationExtensionLockConditional(relation, ExclusiveLock)
+			!= LOCKACQUIRE_OK)
+		{
+			LockRelationForExtension(relation, ExclusiveLock);
+
+			if (use_fsm)
+			{
+				if (lastValidBlock != InvalidBlockNumber)
+				{
+					/*
+					 * Here we are calling RecordAndGetPageWithFreeSpace
+					 * instead of GetPageWithFreeSpace, because other backend
+					 * who have got the lock might have added extra blocks
+					 * in the FSM and its possible that FSM tree might not
+					 * have been updated so far and we will not get the page by
+					 * searching from top using GetPageWithFreeSpace, so we
+					 * need to search the leaf page directly using our
+					 * last valid target block.
+					 */
+					targetBlock = RecordAndGetPageWithFreeSpace(relation,
+														lastValidBlock,
+														pageFreeSpace,
+														len + saveFreeSpace);
+				}
+
+				/* Other waiter has extended the block for us */
+				if (targetBlock != InvalidBlockNumber)
+				{
+					UnlockRelationForExtension(relation, ExclusiveLock);
+					goto loop;
+				}
+
+				RelationAddExtraBlocks(relation, bistate);
+			}
+		}
+	}
 
 	/*
-	 * Now acquire lock on the new page.
+	 * For all case we need to add at least one block to satisfy our
+	 * own request.
 	 */
-	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+	buffer = RelationAddOneBlock(relation, otherBuffer, bistate);
 
 	/*
 	 * Release the file-extension lock; it's now OK for someone else to extend
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 9d16afb..8967d1b 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -341,6 +341,40 @@ LockRelationForExtension(Relation relation, LOCKMODE lockmode)
 }
 
 /*
+ *		RelationExtensionLockConditional
+ *
+ * Same as LockRelationForExtension except it will not wait on the lock.
+ */
+LockAcquireResult
+RelationExtensionLockConditional(Relation relation, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockAcquire(&tag, lockmode, false, true);
+}
+
+/*
+ *		RelationExtensionLockWaiterCount
+ *
+ * Count the lock requester for the RelationExtension lock.
+ */
+int
+RelationExtensionLockWaiterCount(Relation relation)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockWaiterCount(&tag);
+}
+
+/*
  *		UnlockRelationForExtension
  */
 void
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index a458c68..cf723ae 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -4380,3 +4380,47 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	LockRelease(&tag, ShareLock, false);
 	return true;
 }
+
+/*
+ * LockWaiterCount
+ *
+ * Find the number of lock requester on this locktag
+ */
+int
+LockWaiterCount(const LOCKTAG *locktag)
+{
+	LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
+	LOCALLOCKTAG localtag;
+	LOCALLOCK  *locallock;
+	bool		found;
+	uint32		hashcode;
+	LWLock		*partitionLock;
+	int			waiters = 0;
+
+	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
+		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
+
+	/*
+	 * Find a LOCALLOCK entry for this lock and lockmode
+	 */
+	MemSet(&localtag, 0, sizeof(localtag));		/* must clear padding */
+	localtag.lock = *locktag;
+	localtag.mode = ExclusiveLock;
+
+	locallock = (LOCALLOCK *) hash_search(LockMethodLocalHash,
+										(void *) &localtag,
+										HASH_FIND, &found);
+
+
+	hashcode = locallock->hashcode;
+
+	partitionLock = LockHashPartitionLock(hashcode);
+	LWLockAcquire(partitionLock, LW_EXCLUSIVE);
+
+	if (found)
+		waiters = locallock->lock->nRequested;
+
+	LWLockRelease(partitionLock);
+
+	return waiters;
+}
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index e9d41bf..ce9c6aa 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -101,4 +101,7 @@ extern void UnlockSharedObjectForSession(Oid classid, Oid objid, uint16 objsubid
 /* Describe a locktag for error messages */
 extern void DescribeLockTag(StringInfo buf, const LOCKTAG *tag);
 
+extern LockAcquireResult RelationExtensionLockConditional(Relation relation, LOCKMODE lockmode);
+extern int RelationExtensionLockWaiterCount(Relation relation);
+
 #endif   /* LMGR_H */
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index 788d50a..3fd74fb 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -572,6 +572,8 @@ extern void RememberSimpleDeadLock(PGPROC *proc1,
 					   PGPROC *proc2);
 extern void InitDeadLockChecking(void);
 
+extern int LockWaiterCount(const LOCKTAG *locktag);
+
 #ifdef LOCK_DEBUG
 extern void DumpLocks(PGPROC *proc);
 extern void DumpAllLocks(void);

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..3ab911f 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -169,6 +169,55 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 }
 
 /*
+ * Extend a relation by multiple blocks to avoid future contention on the
+ * relation extension lock.  Our goal is to pre-extend the relation by an
+ * amount which ramps up as the degree of contention ramps up, but limiting
+ * the result to some sane overall value.
+ */
+static void
+RelationAddExtraBlocks(Relation relation, BulkInsertState bistate)
+{
+	Page		page;
+	Size		freespace;
+	BlockNumber blockNum;
+	int			extraBlocks = 0;
+	int			lockWaiters = 0;
+	Buffer		buffer;
+
+	/*
+	 * We use the length of the lock wait queue to judge how much to extend.
+	 * It might seem like multiplying the number of lock waiters by as much
+	 * as 20 is too aggressive, but benchmarking revealed that smaller numbers
+	 * were insufficient.  512 is just an arbitrary cap to prevent pathological
+	 * results (and excessive wasted disk space).
+	 */
+	lockWaiters = RelationExtensionLockWaiterCount(relation);
+	extraBlocks = Min(512, lockWaiters * 20);
+
+	while (extraBlocks-- >= 0)
+	{
+		/* Ouch - an unnecessary lseek() each time through the loop! */
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+		/* Extend by one page. */
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+		page = BufferGetPage(buffer);
+		PageInit(page, BufferGetPageSize(buffer), 0);
+		freespace = PageGetHeapFreeSpace(page);
+		MarkBufferDirty(buffer);
+		blockNum = BufferGetBlockNumber(buffer);
+		UnlockReleaseBuffer(buffer);
+
+		/*
+		 * Put the page in the freespace map so other backends can find it.
+		 * This is what will keep those other backends from also queueing up
+		 * on the relation extension lock.
+		 */
+		RecordPageWithFreeSpace(relation, blockNum, freespace);
+	}
+}
+
+/*
  * RelationGetBufferForTuple
  *
  *	Returns pinned and exclusive-locked buffer of a page in given relation
@@ -233,10 +282,11 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	bool		use_fsm = !(options & HEAP_INSERT_SKIP_FSM);
 	Buffer		buffer = InvalidBuffer;
 	Page		page;
-	Size		pageFreeSpace,
-				saveFreeSpace;
+	Size		pageFreeSpace = 0,
+				saveFreeSpace = 0;
 	BlockNumber targetBlock,
-				otherBlock;
+				otherBlock,
+				lastValidBlock = InvalidBlockNumber;
 	bool		needLock;
 
 	len = MAXALIGN(len);		/* be conservative */
@@ -308,6 +358,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		}
 	}
 
+loop:
 	while (targetBlock != InvalidBlockNumber)
 	{
 		/*
@@ -388,6 +439,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 								 otherBlock, targetBlock, vmbuffer_other,
 								 vmbuffer);
 
+		lastValidBlock = targetBlock;
+
 		/*
 		 * Now we can check to see if there's enough free space here. If so,
 		 * we're done.
@@ -440,10 +493,60 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	 */
 	needLock = !RELATION_IS_LOCAL(relation);
 
+	/*
+	 * If we need the lock but are not able to acquire it immediately, we'll
+	 * consider extending the relation by multiple blocks at a time to manage
+	 * contention on the relation extension lock.  However, this only makes
+	 * sense if we're using the FSM; otherwise, there's no point.
+	 */
 	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
+	{
+		if (use_fsm)
+			LockRelationForExtension(relation, ExclusiveLock);
+		else if (!ConditionLockRelationForExtension(relation, ExclusiveLock))
+		{
+			/* Couldn't get the lock immmediately; wait for it. */
+			LockRelationForExtension(relation, ExclusiveLock);
+
+			if (lastValidBlock != InvalidBlockNumber)
+			{
+				/*
+				 * Here we are calling RecordAndGetPageWithFreeSpace
+				 * instead of GetPageWithFreeSpace, because other backend
+				 * who have got the lock might have added extra blocks in
+				 * the FSM and its possible that FSM tree might not have
+				 * been updated so far and we will not get the page by
+				 * searching from top using GetPageWithFreeSpace, so we
+				 * need to search the leaf page directly using our last
+				 * valid target block.
+				 *
+				 * XXX. I don't understand what is happening here. -RMH
+				 */
+				targetBlock = RecordAndGetPageWithFreeSpace(relation,
+														  lastValidBlock,
+															pageFreeSpace,
+													len + saveFreeSpace);
+			}
+
+			/*
+			 * If some other waiter has already extended the relation, we
+			 * don't need to do so; just use the existing freespace.
+			 */
+			if (targetBlock != InvalidBlockNumber)
+			{
+				UnlockRelationForExtension(relation, ExclusiveLock);
+				goto loop;
+			}
+
+			/* Time to bulk-extend. */
+			RelationAddExtraBlocks(relation, bistate);
+		}
+	}
 
 	/*
+	 * In addition to whatever extension we performed above, we always add
+	 * at least one block to satisfy our own request.
+	 *
 	 * XXX This does an lseek - rather expensive - but at the moment it is the
 	 * only way to accurately determine how many blocks are in a relation.  Is
 	 * it worth keeping an accurate file length in shared memory someplace,
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 0632fc0..7e04137 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -341,6 +341,41 @@ LockRelationForExtension(Relation relation, LOCKMODE lockmode)
 }
 
 /*
+ *		ConditionLockRelationForExtension
+ *
+ * As above, but only lock if we can get the lock without blocking.
+ * Returns TRUE iff the lock was acquired.
+ */
+bool
+ConditionLockRelationForExtension(Relation relation, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return (LockAcquire(&tag, lockmode, false, true) != LOCKACQUIRE_NOT_AVAIL);
+}
+
+/*
+ *		RelationExtensionLockWaiterCount
+ *
+ * Count the lock requester for the RelationExtension lock.
+ */
+int
+RelationExtensionLockWaiterCount(Relation relation)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockWaiterCount(&tag);
+}
+
+/*
  *		UnlockRelationForExtension
  */
 void
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index b30b7b1..353f705 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -4380,3 +4380,40 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	LockRelease(&tag, ShareLock, false);
 	return true;
 }
+
+/*
+ * LockWaiterCount
+ *
+ * Find the number of lock requester on this locktag
+ */
+int
+LockWaiterCount(const LOCKTAG *locktag)
+{
+	LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
+	LOCK	   *lock;
+	bool		found;
+	uint32		hashcode;
+	LWLock		*partitionLock;
+	int			waiters = 0;
+
+	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
+		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
+
+	hashcode = LockTagHashCode(locktag);
+	partitionLock = LockHashPartitionLock(hashcode);
+	LWLockAcquire(partitionLock, LW_EXCLUSIVE);
+
+	lock = (LOCK *) hash_search_with_hash_value(LockMethodLockHash,
+												(const void *) locktag,
+												hashcode,
+												HASH_FIND,
+												&found);
+	if (found)
+	{
+		Assert(lock != NULL);
+		waiters = lock->nRequested;
+	}
+	LWLockRelease(partitionLock);
+
+	return waiters;
+}
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index 975b6f8..4460756 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -53,6 +53,9 @@ extern void UnlockRelationIdForSession(LockRelId *relid, LOCKMODE lockmode);
 /* Lock a relation for extension */
 extern void LockRelationForExtension(Relation relation, LOCKMODE lockmode);
 extern void UnlockRelationForExtension(Relation relation, LOCKMODE lockmode);
+extern bool ConditionLockRelationForExtension(Relation relation,
+								  LOCKMODE lockmode);
+extern int	RelationExtensionLockWaiterCount(Relation relation);
 
 /* Lock a page (currently only used within indexes) */
 extern void LockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode);
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index b26427d..9c08679 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -574,6 +574,8 @@ extern void RememberSimpleDeadLock(PGPROC *proc1,
 					   PGPROC *proc2);
 extern void InitDeadLockChecking(void);
 
+extern int LockWaiterCount(const LOCKTAG *locktag);
+
 #ifdef LOCK_DEBUG
 extern void DumpLocks(PGPROC *proc);
 extern void DumpAllLocks(void);

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..3b46c4f 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -169,6 +169,55 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 }
 
 /*
+ * Extend a relation by multiple blocks to avoid future contention on the
+ * relation extension lock.  Our goal is to pre-extend the relation by an
+ * amount which ramps up as the degree of contention ramps up, but limiting
+ * the result to some sane overall value.
+ */
+static void
+RelationAddExtraBlocks(Relation relation, BulkInsertState bistate)
+{
+	Page		page;
+	Size		freespace;
+	BlockNumber blockNum;
+	int			extraBlocks = 0;
+	int			lockWaiters = 0;
+	Buffer		buffer;
+
+	/*
+	 * We use the length of the lock wait queue to judge how much to extend.
+	 * It might seem like multiplying the number of lock waiters by as much
+	 * as 20 is too aggressive, but benchmarking revealed that smaller numbers
+	 * were insufficient.  512 is just an arbitrary cap to prevent pathological
+	 * results (and excessive wasted disk space).
+	 */
+	lockWaiters = RelationExtensionLockWaiterCount(relation);
+	extraBlocks = Min(512, lockWaiters * 20);
+
+	while (extraBlocks-- >= 0)
+	{
+		/* Ouch - an unnecessary lseek() each time through the loop! */
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+		/* Extend by one page. */
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+		page = BufferGetPage(buffer);
+		PageInit(page, BufferGetPageSize(buffer), 0);
+		freespace = PageGetHeapFreeSpace(page);
+		MarkBufferDirty(buffer);
+		blockNum = BufferGetBlockNumber(buffer);
+		UnlockReleaseBuffer(buffer);
+
+		/*
+		 * Put the page in the freespace map so other backends can find it.
+		 * This is what will keep those other backends from also queueing up
+		 * on the relation extension lock.
+		 */
+		RecordPageWithFreeSpace(relation, blockNum, freespace);
+	}
+}
+
+/*
  * RelationGetBufferForTuple
  *
  *	Returns pinned and exclusive-locked buffer of a page in given relation
@@ -233,10 +282,11 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	bool		use_fsm = !(options & HEAP_INSERT_SKIP_FSM);
 	Buffer		buffer = InvalidBuffer;
 	Page		page;
-	Size		pageFreeSpace,
-				saveFreeSpace;
+	Size		pageFreeSpace = 0,
+				saveFreeSpace = 0;
 	BlockNumber targetBlock,
-				otherBlock;
+				otherBlock,
+				lastValidBlock = InvalidBlockNumber;
 	bool		needLock;
 
 	len = MAXALIGN(len);		/* be conservative */
@@ -308,6 +358,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		}
 	}
 
+loop:
 	while (targetBlock != InvalidBlockNumber)
 	{
 		/*
@@ -388,6 +439,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 								 otherBlock, targetBlock, vmbuffer_other,
 								 vmbuffer);
 
+		lastValidBlock = targetBlock;
+
 		/*
 		 * Now we can check to see if there's enough free space here. If so,
 		 * we're done.
@@ -440,10 +493,57 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	 */
 	needLock = !RELATION_IS_LOCAL(relation);
 
+	/*
+	 * If we need the lock but are not able to acquire it immediately, we'll
+	 * consider extending the relation by multiple blocks at a time to manage
+	 * contention on the relation extension lock.  However, this only makes
+	 * sense if we're using the FSM; otherwise, there's no point.
+	 */
 	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
+	{
+		if (!use_fsm)
+			LockRelationForExtension(relation, ExclusiveLock);
+		else if (!ConditionLockRelationForExtension(relation, ExclusiveLock))
+		{
+			/* Couldn't get the lock immmediately; wait for it. */
+			LockRelationForExtension(relation, ExclusiveLock);
+
+			if (lastValidBlock != InvalidBlockNumber)
+			{
+				/*
+				 * Here we are calling GetPageWithFreeSpaceUsingOldPage
+				 * instead of GetPageWithFreeSpace, because other backend
+				 * who have got the lock might have added extra blocks in
+				 * the FSM and its possible that free space information
+				 * is not yet propagated up till root node (it will be
+				 * updated during vacuum).
+				 * So directly start search from leaf level where we ended
+				 * the search last time.
+				 */
+				targetBlock = GetPageWithFreeSpaceUsingOldPage(relation,
+														lastValidBlock,
+														len + saveFreeSpace);
+			}
+
+			/*
+			 * If some other waiter has already extended the relation, we
+			 * don't need to do so; just use the existing freespace.
+			 */
+			if (targetBlock != InvalidBlockNumber)
+			{
+				UnlockRelationForExtension(relation, ExclusiveLock);
+				goto loop;
+			}
+
+			/* Time to bulk-extend. */
+			RelationAddExtraBlocks(relation, bistate);
+		}
+	}
 
 	/*
+	 * In addition to whatever extension we performed above, we always add
+	 * at least one block to satisfy our own request.
+	 *
 	 * XXX This does an lseek - rather expensive - but at the moment it is the
 	 * only way to accurately determine how many blocks are in a relation.  Is
 	 * it worth keeping an accurate file length in shared memory someplace,
diff --git a/src/backend/storage/freespace/freespace.c b/src/backend/storage/freespace/freespace.c
index 2631080..ad94b08 100644
--- a/src/backend/storage/freespace/freespace.c
+++ b/src/backend/storage/freespace/freespace.c
@@ -109,6 +109,7 @@ static int fsm_set_and_search(Relation rel, FSMAddress addr, uint16 slot,
 				   uint8 newValue, uint8 minValue);
 static BlockNumber fsm_search(Relation rel, uint8 min_cat);
 static uint8 fsm_vacuum_page(Relation rel, FSMAddress addr, bool *eof);
+static int fsm_search_from_addr(Relation rel, FSMAddress addr, uint8 minValue);
 
 
 /******** Public API ********/
@@ -135,6 +136,37 @@ GetPageWithFreeSpace(Relation rel, Size spaceNeeded)
 }
 
 /*
+ * 		GetPageWithFreeSpaceUsingOldPage
+ *
+ * As above, but start the search from oldPage instead of staring from root
+ * So that, we can find the appropriate page in cases where free block is
+ * added to FSM but not yet updated up till root.
+ */
+BlockNumber
+GetPageWithFreeSpaceUsingOldPage(Relation rel, BlockNumber oldPage,
+								Size spaceNeeded)
+{
+	int			search_cat = fsm_space_needed_to_cat(spaceNeeded);
+	FSMAddress	addr;
+	uint16		slot;
+	int			search_slot;
+
+	/* Get the location of the FSM byte representing the heap block */
+	addr = fsm_get_location(oldPage, &slot);
+
+	search_slot = fsm_search_from_addr(rel, addr, search_cat);
+
+	/*
+	 * If fsm_set_and_search found a suitable new block, return that.
+	 * Otherwise, search as usual.
+	 */
+	if (search_slot != -1)
+		return fsm_get_heap_blk(addr, search_slot);
+	else
+		return fsm_search(rel, search_cat);
+}
+
+/*
  * RecordAndGetPageWithFreeSpace - update info about a page and try again.
  *
  * We provide this combo form to save some locking overhead, compared to
@@ -634,6 +666,35 @@ fsm_set_and_search(Relation rel, FSMAddress addr, uint16 slot,
 }
 
 /*
+ * Search the the fsm tree for for free space > minValue
+ * It will start the search from given addr, and will be used for searching
+ * the required page in cases where vacuum have not yet updated the FSM tree
+ * till root level.
+ * If one is found, its slot number is returned, -1 otherwise.
+ */
+static int
+fsm_search_from_addr(Relation rel, FSMAddress addr, uint8 minValue)
+{
+	Buffer		buf;
+	int			newslot = -1;
+
+	buf = fsm_readbuf(rel, addr, true);
+	LockBuffer(buf, BUFFER_LOCK_SHARE);
+
+	if (minValue != 0)
+	{
+		/* Search while we still hold the lock */
+		newslot = fsm_search_avail(buf, minValue,
+								   addr.level == FSM_BOTTOM_LEVEL,
+								   false);
+	}
+
+	UnlockReleaseBuffer(buf);
+
+	return newslot;
+}
+
+/*
  * Search the tree for a heap page with at least min_cat of free space
  */
 static BlockNumber
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 0632fc0..7e04137 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -341,6 +341,41 @@ LockRelationForExtension(Relation relation, LOCKMODE lockmode)
 }
 
 /*
+ *		ConditionLockRelationForExtension
+ *
+ * As above, but only lock if we can get the lock without blocking.
+ * Returns TRUE iff the lock was acquired.
+ */
+bool
+ConditionLockRelationForExtension(Relation relation, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return (LockAcquire(&tag, lockmode, false, true) != LOCKACQUIRE_NOT_AVAIL);
+}
+
+/*
+ *		RelationExtensionLockWaiterCount
+ *
+ * Count the lock requester for the RelationExtension lock.
+ */
+int
+RelationExtensionLockWaiterCount(Relation relation)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockWaiterCount(&tag);
+}
+
+/*
  *		UnlockRelationForExtension
  */
 void
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index b30b7b1..353f705 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -4380,3 +4380,40 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	LockRelease(&tag, ShareLock, false);
 	return true;
 }
+
+/*
+ * LockWaiterCount
+ *
+ * Find the number of lock requester on this locktag
+ */
+int
+LockWaiterCount(const LOCKTAG *locktag)
+{
+	LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
+	LOCK	   *lock;
+	bool		found;
+	uint32		hashcode;
+	LWLock		*partitionLock;
+	int			waiters = 0;
+
+	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
+		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
+
+	hashcode = LockTagHashCode(locktag);
+	partitionLock = LockHashPartitionLock(hashcode);
+	LWLockAcquire(partitionLock, LW_EXCLUSIVE);
+
+	lock = (LOCK *) hash_search_with_hash_value(LockMethodLockHash,
+												(const void *) locktag,
+												hashcode,
+												HASH_FIND,
+												&found);
+	if (found)
+	{
+		Assert(lock != NULL);
+		waiters = lock->nRequested;
+	}
+	LWLockRelease(partitionLock);
+
+	return waiters;
+}
diff --git a/src/include/storage/freespace.h b/src/include/storage/freespace.h
index 19dcb8d..0e67e47 100644
--- a/src/include/storage/freespace.h
+++ b/src/include/storage/freespace.h
@@ -32,5 +32,9 @@ extern void XLogRecordPageWithFreeSpace(RelFileNode rnode, BlockNumber heapBlk,
 
 extern void FreeSpaceMapTruncateRel(Relation rel, BlockNumber nblocks);
 extern void FreeSpaceMapVacuum(Relation rel);
+extern BlockNumber GetPageWithFreeSpaceUsingOldPage(Relation rel,
+								BlockNumber oldPage,
+								Size spaceNeeded);
+
 
 #endif   /* FREESPACE_H_ */
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index 975b6f8..4460756 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -53,6 +53,9 @@ extern void UnlockRelationIdForSession(LockRelId *relid, LOCKMODE lockmode);
 /* Lock a relation for extension */
 extern void LockRelationForExtension(Relation relation, LOCKMODE lockmode);
 extern void UnlockRelationForExtension(Relation relation, LOCKMODE lockmode);
+extern bool ConditionLockRelationForExtension(Relation relation,
+								  LOCKMODE lockmode);
+extern int	RelationExtensionLockWaiterCount(Relation relation);
 
 /* Lock a page (currently only used within indexes) */
 extern void LockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode);
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index b26427d..9c08679 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -574,6 +574,8 @@ extern void RememberSimpleDeadLock(PGPROC *proc1,
 					   PGPROC *proc2);
 extern void InitDeadLockChecking(void);
 
+extern int LockWaiterCount(const LOCKTAG *locktag);
+
 #ifdef LOCK_DEBUG
 extern void DumpLocks(PGPROC *proc);
 extern void DumpAllLocks(void);

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..d3608c6 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -169,6 +169,55 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 }
 
 /*
+ * Extend a relation by multiple blocks to avoid future contention on the
+ * relation extension lock.  Our goal is to pre-extend the relation by an
+ * amount which ramps up as the degree of contention ramps up, but limiting
+ * the result to some sane overall value.
+ */
+static void
+RelationAddExtraBlocks(Relation relation, BulkInsertState bistate)
+{
+	Page		page;
+	Size		freespace;
+	BlockNumber blockNum;
+	int			extraBlocks = 0;
+	int			lockWaiters = 0;
+	Buffer		buffer;
+
+	/*
+	 * We use the length of the lock wait queue to judge how much to extend.
+	 * It might seem like multiplying the number of lock waiters by as much
+	 * as 20 is too aggressive, but benchmarking revealed that smaller numbers
+	 * were insufficient.  512 is just an arbitrary cap to prevent pathological
+	 * results (and excessive wasted disk space).
+	 */
+	lockWaiters = RelationExtensionLockWaiterCount(relation);
+	extraBlocks = Min(512, lockWaiters * 20);
+
+	while (extraBlocks-- >= 0)
+	{
+		/* Ouch - an unnecessary lseek() each time through the loop! */
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+		/* Extend by one page. */
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+		page = BufferGetPage(buffer);
+		PageInit(page, BufferGetPageSize(buffer), 0);
+		freespace = PageGetHeapFreeSpace(page);
+		MarkBufferDirty(buffer);
+		blockNum = BufferGetBlockNumber(buffer);
+		UnlockReleaseBuffer(buffer);
+
+		/*
+		 * Put the page in the freespace map so other backends can find it.
+		 * This is what will keep those other backends from also queueing up
+		 * on the relation extension lock.
+		 */
+		RecordPageWithFreeSpace(relation, blockNum, freespace);
+	}
+}
+
+/*
  * RelationGetBufferForTuple
  *
  *	Returns pinned and exclusive-locked buffer of a page in given relation
@@ -233,10 +282,11 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	bool		use_fsm = !(options & HEAP_INSERT_SKIP_FSM);
 	Buffer		buffer = InvalidBuffer;
 	Page		page;
-	Size		pageFreeSpace,
-				saveFreeSpace;
+	Size		pageFreeSpace = 0,
+				saveFreeSpace = 0;
 	BlockNumber targetBlock,
-				otherBlock;
+				otherBlock,
+				lastValidBlock = InvalidBlockNumber;
 	bool		needLock;
 
 	len = MAXALIGN(len);		/* be conservative */
@@ -308,6 +358,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		}
 	}
 
+loop:
 	while (targetBlock != InvalidBlockNumber)
 	{
 		/*
@@ -388,6 +439,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 								 otherBlock, targetBlock, vmbuffer_other,
 								 vmbuffer);
 
+		lastValidBlock = targetBlock;
+
 		/*
 		 * Now we can check to see if there's enough free space here. If so,
 		 * we're done.
@@ -440,10 +493,57 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	 */
 	needLock = !RELATION_IS_LOCAL(relation);
 
+	/*
+	 * If we need the lock but are not able to acquire it immediately, we'll
+	 * consider extending the relation by multiple blocks at a time to manage
+	 * contention on the relation extension lock.  However, this only makes
+	 * sense if we're using the FSM; otherwise, there's no point.
+	 */
 	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
+	{
+		if (!use_fsm)
+			LockRelationForExtension(relation, ExclusiveLock);
+		else if (!ConditionLockRelationForExtension(relation, ExclusiveLock))
+		{
+			/* Couldn't get the lock immmediately; wait for it. */
+			LockRelationForExtension(relation, ExclusiveLock);
+
+			if (lastValidBlock != InvalidBlockNumber)
+			{
+				/*
+				 * Here we are calling GetPageWithFreeSpaceUsingOldPage
+				 * instead of GetPageWithFreeSpace, because other backend
+				 * who have got the lock might have added extra blocks in
+				 * the FSM and its possible that free space information
+				 * is not yet propagated up till root node (it will be
+				 * updated during vacuum).
+				 * So directly start search from leaf level where we ended
+				 * the search last time.
+				 */
+				targetBlock = GetPageWithFreeSpaceExtended(relation,
+														lastValidBlock,
+														len + saveFreeSpace);
+			}
+
+			/*
+			 * If some other waiter has already extended the relation, we
+			 * don't need to do so; just use the existing freespace.
+			 */
+			if (targetBlock != InvalidBlockNumber)
+			{
+				UnlockRelationForExtension(relation, ExclusiveLock);
+				goto loop;
+			}
+
+			/* Time to bulk-extend. */
+			RelationAddExtraBlocks(relation, bistate);
+		}
+	}
 
 	/*
+	 * In addition to whatever extension we performed above, we always add
+	 * at least one block to satisfy our own request.
+	 *
 	 * XXX This does an lseek - rather expensive - but at the moment it is the
 	 * only way to accurately determine how many blocks are in a relation.  Is
 	 * it worth keeping an accurate file length in shared memory someplace,
diff --git a/src/backend/storage/freespace/freespace.c b/src/backend/storage/freespace/freespace.c
index 2631080..980651e 100644
--- a/src/backend/storage/freespace/freespace.c
+++ b/src/backend/storage/freespace/freespace.c
@@ -109,6 +109,7 @@ static int fsm_set_and_search(Relation rel, FSMAddress addr, uint16 slot,
 				   uint8 newValue, uint8 minValue);
 static BlockNumber fsm_search(Relation rel, uint8 min_cat);
 static uint8 fsm_vacuum_page(Relation rel, FSMAddress addr, bool *eof);
+static int fsm_search_from_addr(Relation rel, FSMAddress addr, uint8 minValue);
 
 
 /******** Public API ********/
@@ -129,9 +130,46 @@ static uint8 fsm_vacuum_page(Relation rel, FSMAddress addr, bool *eof);
 BlockNumber
 GetPageWithFreeSpace(Relation rel, Size spaceNeeded)
 {
-	uint8		min_cat = fsm_space_needed_to_cat(spaceNeeded);
+	/*
+	 * Call GetPageWithFreeSpaceExtended with InvalidBlockNumber so that
+	 * it will search the FSM tree from the root
+	 */
+	return GetPageWithFreeSpaceExtended(rel, InvalidBlockNumber, spaceNeeded);
+}
+
+/*
+ * 		GetPageWithFreeSpaceExtended
+ *
+ * As above, but start the search from oldPage instead of staring from root
+ * So that, we can find the appropriate page in cases where free block is
+ * added to FSM but not yet updated up till root. If oldpage is Invalid
+ * then start the search from root.
+ */
+BlockNumber
+GetPageWithFreeSpaceExtended(Relation rel, BlockNumber oldPage,
+								Size spaceNeeded)
+{
+	int			search_cat = fsm_space_needed_to_cat(spaceNeeded);
+	FSMAddress	addr;
+	uint16		slot;
+	int			search_slot = -1;
+
+	if (oldPage != InvalidBlockNumber)
+	{
+		/* Get the location of the FSM byte representing the heap block */
+		addr = fsm_get_location(oldPage, &slot);
+
+		search_slot = fsm_search_from_addr(rel, addr, search_cat);
+	}
 
-	return fsm_search(rel, min_cat);
+	/*
+	 * If fsm_search_from_addr found a suitable new block, return that.
+	 * Otherwise, search as usual.
+	 */
+	if (search_slot != -1)
+		return fsm_get_heap_blk(addr, search_slot);
+	else
+		return fsm_search(rel, search_cat);
 }
 
 /*
@@ -634,6 +672,34 @@ fsm_set_and_search(Relation rel, FSMAddress addr, uint16 slot,
 }
 
 /*
+ * Search the the fsm tree for for free space > minValue
+ * It will start the search from given addr, and will be used for searching
+ * the required page in cases where vacuum have not yet updated the FSM tree
+ * till root level.
+ * If one is found, its slot number is returned, -1 otherwise.
+ */
+static int
+fsm_search_from_addr(Relation rel, FSMAddress addr, uint8 minValue)
+{
+	Buffer		buf;
+	int			newslot = -1;
+
+	buf = fsm_readbuf(rel, addr, true);
+	LockBuffer(buf, BUFFER_LOCK_SHARE);
+
+	Assert(minValue != 0);
+
+	/* Search while we still hold the lock */
+	newslot = fsm_search_avail(buf, minValue,
+							addr.level == FSM_BOTTOM_LEVEL,
+							false);
+
+	UnlockReleaseBuffer(buf);
+
+	return newslot;
+}
+
+/*
  * Search the tree for a heap page with at least min_cat of free space
  */
 static BlockNumber
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 0632fc0..7e04137 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -341,6 +341,41 @@ LockRelationForExtension(Relation relation, LOCKMODE lockmode)
 }
 
 /*
+ *		ConditionLockRelationForExtension
+ *
+ * As above, but only lock if we can get the lock without blocking.
+ * Returns TRUE iff the lock was acquired.
+ */
+bool
+ConditionLockRelationForExtension(Relation relation, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return (LockAcquire(&tag, lockmode, false, true) != LOCKACQUIRE_NOT_AVAIL);
+}
+
+/*
+ *		RelationExtensionLockWaiterCount
+ *
+ * Count the lock requester for the RelationExtension lock.
+ */
+int
+RelationExtensionLockWaiterCount(Relation relation)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockWaiterCount(&tag);
+}
+
+/*
  *		UnlockRelationForExtension
  */
 void
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index b30b7b1..353f705 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -4380,3 +4380,40 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	LockRelease(&tag, ShareLock, false);
 	return true;
 }
+
+/*
+ * LockWaiterCount
+ *
+ * Find the number of lock requester on this locktag
+ */
+int
+LockWaiterCount(const LOCKTAG *locktag)
+{
+	LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
+	LOCK	   *lock;
+	bool		found;
+	uint32		hashcode;
+	LWLock		*partitionLock;
+	int			waiters = 0;
+
+	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
+		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
+
+	hashcode = LockTagHashCode(locktag);
+	partitionLock = LockHashPartitionLock(hashcode);
+	LWLockAcquire(partitionLock, LW_EXCLUSIVE);
+
+	lock = (LOCK *) hash_search_with_hash_value(LockMethodLockHash,
+												(const void *) locktag,
+												hashcode,
+												HASH_FIND,
+												&found);
+	if (found)
+	{
+		Assert(lock != NULL);
+		waiters = lock->nRequested;
+	}
+	LWLockRelease(partitionLock);
+
+	return waiters;
+}
diff --git a/src/include/storage/freespace.h b/src/include/storage/freespace.h
index 19dcb8d..27cb971 100644
--- a/src/include/storage/freespace.h
+++ b/src/include/storage/freespace.h
@@ -32,5 +32,9 @@ extern void XLogRecordPageWithFreeSpace(RelFileNode rnode, BlockNumber heapBlk,
 
 extern void FreeSpaceMapTruncateRel(Relation rel, BlockNumber nblocks);
 extern void FreeSpaceMapVacuum(Relation rel);
+extern BlockNumber GetPageWithFreeSpaceExtended(Relation rel,
+								BlockNumber oldPage,
+								Size spaceNeeded);
+
 
 #endif   /* FREESPACE_H_ */
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index 975b6f8..4460756 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -53,6 +53,9 @@ extern void UnlockRelationIdForSession(LockRelId *relid, LOCKMODE lockmode);
 /* Lock a relation for extension */
 extern void LockRelationForExtension(Relation relation, LOCKMODE lockmode);
 extern void UnlockRelationForExtension(Relation relation, LOCKMODE lockmode);
+extern bool ConditionLockRelationForExtension(Relation relation,
+								  LOCKMODE lockmode);
+extern int	RelationExtensionLockWaiterCount(Relation relation);
 
 /* Lock a page (currently only used within indexes) */
 extern void LockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode);
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index b26427d..9c08679 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -574,6 +574,8 @@ extern void RememberSimpleDeadLock(PGPROC *proc1,
 					   PGPROC *proc2);
 extern void InitDeadLockChecking(void);
 
+extern int LockWaiterCount(const LOCKTAG *locktag);
+
 #ifdef LOCK_DEBUG
 extern void DumpLocks(PGPROC *proc);
 extern void DumpAllLocks(void);

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..9ac1eb7 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -169,6 +169,62 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 }
 
 /*
+ * Extend a relation by multiple blocks to avoid future contention on the
+ * relation extension lock.  Our goal is to pre-extend the relation by an
+ * amount which ramps up as the degree of contention ramps up, but limiting
+ * the result to some sane overall value.
+ */
+static void
+RelationAddExtraBlocks(Relation relation, BulkInsertState bistate)
+{
+	Page		page;
+	BlockNumber blockNum,
+				firstBlock,
+				lastBlock;
+	int			extraBlocks = 0;
+	int			lockWaiters = 0;
+	Buffer		buffer;
+
+	/*
+	 * We use the length of the lock wait queue to judge how much to extend.
+	 * It might seem like multiplying the number of lock waiters by as much
+	 * as 20 is too aggressive, but benchmarking revealed that smaller numbers
+	 * were insufficient.  512 is just an arbitrary cap to prevent pathological
+	 * results (and excessive wasted disk space).
+	 */
+	lockWaiters = RelationExtensionLockWaiterCount(relation);
+	extraBlocks = Min(512, lockWaiters * 20);
+
+	firstBlock = lastBlock = InvalidBlockNumber;
+
+	while (extraBlocks-- >= 0)
+	{
+		/* Ouch - an unnecessary lseek() each time through the loop! */
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+		/* Extend by one page. */
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+		page = BufferGetPage(buffer);
+		PageInit(page, BufferGetPageSize(buffer), 0);
+		MarkBufferDirty(buffer);
+		blockNum = BufferGetBlockNumber(buffer);
+		UnlockReleaseBuffer(buffer);
+
+		if (firstBlock == InvalidBlockNumber)
+			firstBlock = blockNum;
+
+		lastBlock = blockNum;
+	}
+
+	/*
+	 * Put the page in the freespace map so other backends can find it.
+	 * This is what will keep those other backends from also queueing up
+	 * on the relation extension lock.
+	 */
+	FreeSpaceMapBulkExtend(relation, firstBlock, lastBlock);
+}
+
+/*
  * RelationGetBufferForTuple
  *
  *	Returns pinned and exclusive-locked buffer of a page in given relation
@@ -233,8 +289,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	bool		use_fsm = !(options & HEAP_INSERT_SKIP_FSM);
 	Buffer		buffer = InvalidBuffer;
 	Page		page;
-	Size		pageFreeSpace,
-				saveFreeSpace;
+	Size		pageFreeSpace = 0,
+				saveFreeSpace = 0;
 	BlockNumber targetBlock,
 				otherBlock;
 	bool		needLock;
@@ -308,6 +364,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		}
 	}
 
+loop:
 	while (targetBlock != InvalidBlockNumber)
 	{
 		/*
@@ -440,10 +497,46 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	 */
 	needLock = !RELATION_IS_LOCAL(relation);
 
+	/*
+	 * If we need the lock but are not able to acquire it immediately, we'll
+	 * consider extending the relation by multiple blocks at a time to manage
+	 * contention on the relation extension lock.  However, this only makes
+	 * sense if we're using the FSM; otherwise, there's no point.
+	 */
 	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
+	{
+		if (!use_fsm)
+			LockRelationForExtension(relation, ExclusiveLock);
+		else if (!ConditionLockRelationForExtension(relation, ExclusiveLock))
+		{
+			/* Couldn't get the lock immediately; wait for it. */
+			LockRelationForExtension(relation, ExclusiveLock);
+
+			/*
+			 * Check if some other backend has extended a block for us while
+			 * we were waiting on the lock.
+			 */
+			targetBlock = GetPageWithFreeSpace(relation, len + saveFreeSpace);
+
+			/*
+			 * If some other waiter has already extended the relation, we
+			 * don't need to do so; just use the existing freespace.
+			 */
+			if (targetBlock != InvalidBlockNumber)
+			{
+				UnlockRelationForExtension(relation, ExclusiveLock);
+				goto loop;
+			}
+
+			/* Time to bulk-extend. */
+			RelationAddExtraBlocks(relation, bistate);
+		}
+	}
 
 	/*
+	 * In addition to whatever extension we performed above, we always add
+	 * at least one block to satisfy our own request.
+	 *
 	 * XXX This does an lseek - rather expensive - but at the moment it is the
 	 * only way to accurately determine how many blocks are in a relation.  Is
 	 * it worth keeping an accurate file length in shared memory someplace,
diff --git a/src/backend/storage/freespace/freespace.c b/src/backend/storage/freespace/freespace.c
index 2631080..5f45891 100644
--- a/src/backend/storage/freespace/freespace.c
+++ b/src/backend/storage/freespace/freespace.c
@@ -109,7 +109,10 @@ static int fsm_set_and_search(Relation rel, FSMAddress addr, uint16 slot,
 				   uint8 newValue, uint8 minValue);
 static BlockNumber fsm_search(Relation rel, uint8 min_cat);
 static uint8 fsm_vacuum_page(Relation rel, FSMAddress addr, bool *eof);
-
+static bool fsm_addr_on_same_page(FSMAddress addr1, FSMAddress addr2);
+static void fsm_set_range(Relation rel, FSMAddress addr, uint16 firstSlot,
+					uint16 lastSlot, uint8 newValue);
+static void fsm_update_recursive(Relation rel, FSMAddress addr, uint8 new_cat);
 
 /******** Public API ********/
 
@@ -189,6 +192,49 @@ RecordPageWithFreeSpace(Relation rel, BlockNumber heapBlk, Size spaceAvail)
 }
 
 /*
+ * Sets all the FSM entries for pages between firstBlock and lastBlock to
+ * BLKSIZE and then bubbles that up to the higher levels of the tree and
+ * all the way to the root.
+ */
+void
+FreeSpaceMapBulkExtend(Relation rel, BlockNumber firstBlock,
+					BlockNumber lastBlock)
+{
+	int			new_cat = fsm_space_avail_to_cat(BLCKSZ);
+	FSMAddress	addr1,
+				addr2;
+	uint16		firstSlot;
+	uint16		lastSlot;
+	uint16		slot;
+	BlockNumber	blockNum;
+
+	blockNum = firstBlock;
+	addr1 = fsm_get_location(firstBlock, &slot);
+
+	while (blockNum < lastBlock)
+	{
+		blockNum++;
+
+		addr2 = fsm_get_location(blockNum, &slot);
+
+		if (!fsm_addr_on_same_page(addr1, addr2)
+			|| blockNum == lastBlock)
+		{
+			/* This BLock is on the next FSM page so update the FSM page */
+			fsm_get_location(firstBlock, &firstSlot);
+			fsm_get_location(blockNum - 1, &lastSlot);
+			fsm_set_range(rel, addr1, firstSlot, lastSlot, new_cat);
+			fsm_update_recursive(rel, addr1, new_cat);
+			/*
+			 * Continue updating FSM till last page, so set the addr1 as new
+			 * Address. and continue search for this page.
+			 */
+			addr1 = addr2;
+		}
+	}
+}
+
+/*
  * XLogRecordPageWithFreeSpace - like RecordPageWithFreeSpace, for use in
  *		WAL replay
  */
@@ -788,3 +834,56 @@ fsm_vacuum_page(Relation rel, FSMAddress addr, bool *eof_p)
 
 	return max_avail;
 }
+
+static bool
+fsm_addr_on_same_page(FSMAddress addr1, FSMAddress addr2)
+{
+	Assert(addr1);
+	Assert(addr2);
+
+	if ((addr1.level != addr2.level)
+		|| (addr1.logpageno != addr2.logpageno))
+		return false;
+
+	return true;
+}
+
+/*
+ * Set value in given FSM page for given slot range.
+ */
+static void
+fsm_set_range(Relation rel, FSMAddress addr, uint16 firstSlot,
+			uint16 lastSlot, uint8 newValue)
+{
+	Buffer		buf;
+	Page		page;
+	int			slot;
+
+	buf = fsm_readbuf(rel, addr, true);
+	LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
+
+	page = BufferGetPage(buf);
+
+	slot = firstSlot;
+
+	for (slot = firstSlot; slot <= lastSlot; slot++)
+		fsm_set_avail(page, slot, newValue);
+
+	MarkBufferDirtyHint(buf, false);
+
+	UnlockReleaseBuffer(buf);
+}
+
+static void
+fsm_update_recursive(Relation rel, FSMAddress addr, uint8 new_cat)
+{
+	uint16		parentslot;
+	FSMAddress	parent;
+
+	if (addr.level == FSM_ROOT_LEVEL)
+		return;
+
+	parent = fsm_get_parent(addr, &parentslot);
+	fsm_set_and_search(rel, parent, parentslot, new_cat, 0);
+	fsm_update_recursive(rel, parent, new_cat);
+}
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 0632fc0..7e04137 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -341,6 +341,41 @@ LockRelationForExtension(Relation relation, LOCKMODE lockmode)
 }
 
 /*
+ *		ConditionLockRelationForExtension
+ *
+ * As above, but only lock if we can get the lock without blocking.
+ * Returns TRUE iff the lock was acquired.
+ */
+bool
+ConditionLockRelationForExtension(Relation relation, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return (LockAcquire(&tag, lockmode, false, true) != LOCKACQUIRE_NOT_AVAIL);
+}
+
+/*
+ *		RelationExtensionLockWaiterCount
+ *
+ * Count the lock requester for the RelationExtension lock.
+ */
+int
+RelationExtensionLockWaiterCount(Relation relation)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockWaiterCount(&tag);
+}
+
+/*
  *		UnlockRelationForExtension
  */
 void
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index b30b7b1..353f705 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -4380,3 +4380,40 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	LockRelease(&tag, ShareLock, false);
 	return true;
 }
+
+/*
+ * LockWaiterCount
+ *
+ * Find the number of lock requester on this locktag
+ */
+int
+LockWaiterCount(const LOCKTAG *locktag)
+{
+	LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
+	LOCK	   *lock;
+	bool		found;
+	uint32		hashcode;
+	LWLock		*partitionLock;
+	int			waiters = 0;
+
+	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
+		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
+
+	hashcode = LockTagHashCode(locktag);
+	partitionLock = LockHashPartitionLock(hashcode);
+	LWLockAcquire(partitionLock, LW_EXCLUSIVE);
+
+	lock = (LOCK *) hash_search_with_hash_value(LockMethodLockHash,
+												(const void *) locktag,
+												hashcode,
+												HASH_FIND,
+												&found);
+	if (found)
+	{
+		Assert(lock != NULL);
+		waiters = lock->nRequested;
+	}
+	LWLockRelease(partitionLock);
+
+	return waiters;
+}
diff --git a/src/include/storage/freespace.h b/src/include/storage/freespace.h
index 19dcb8d..0f2a2be 100644
--- a/src/include/storage/freespace.h
+++ b/src/include/storage/freespace.h
@@ -32,5 +32,7 @@ extern void XLogRecordPageWithFreeSpace(RelFileNode rnode, BlockNumber heapBlk,
 
 extern void FreeSpaceMapTruncateRel(Relation rel, BlockNumber nblocks);
 extern void FreeSpaceMapVacuum(Relation rel);
+extern void FreeSpaceMapBulkExtend(Relation rel, BlockNumber firstBlock,
+							BlockNumber lastBlock);
 
 #endif   /* FREESPACE_H_ */
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index 975b6f8..4460756 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -53,6 +53,9 @@ extern void UnlockRelationIdForSession(LockRelId *relid, LOCKMODE lockmode);
 /* Lock a relation for extension */
 extern void LockRelationForExtension(Relation relation, LOCKMODE lockmode);
 extern void UnlockRelationForExtension(Relation relation, LOCKMODE lockmode);
+extern bool ConditionLockRelationForExtension(Relation relation,
+								  LOCKMODE lockmode);
+extern int	RelationExtensionLockWaiterCount(Relation relation);
 
 /* Lock a page (currently only used within indexes) */
 extern void LockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode);
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index b26427d..9c08679 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -574,6 +574,8 @@ extern void RememberSimpleDeadLock(PGPROC *proc1,
 					   PGPROC *proc2);
 extern void InitDeadLockChecking(void);
 
+extern int LockWaiterCount(const LOCKTAG *locktag);
+
 #ifdef LOCK_DEBUG
 extern void DumpLocks(PGPROC *proc);
 extern void DumpAllLocks(void);

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..7d322f8 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -169,6 +169,62 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 }
 
 /*
+ * Extend a relation by multiple blocks to avoid future contention on the
+ * relation extension lock.  Our goal is to pre-extend the relation by an
+ * amount which ramps up as the degree of contention ramps up, but limiting
+ * the result to some sane overall value.
+ */
+static void
+RelationAddExtraBlocks(Relation relation, BulkInsertState bistate)
+{
+	Page		page;
+	BlockNumber blockNum,
+				firstBlock,
+				lastBlock;
+	int			extraBlocks = 0;
+	int			lockWaiters = 0;
+	Buffer		buffer;
+
+	/*
+	 * We use the length of the lock wait queue to judge how much to extend.
+	 * It might seem like multiplying the number of lock waiters by as much
+	 * as 20 is too aggressive, but benchmarking revealed that smaller numbers
+	 * were insufficient.  512 is just an arbitrary cap to prevent pathological
+	 * results (and excessive wasted disk space).
+	 */
+	lockWaiters = RelationExtensionLockWaiterCount(relation);
+	extraBlocks = Min(512, lockWaiters * 20);
+
+	firstBlock = lastBlock = InvalidBlockNumber;
+
+	while (extraBlocks-- >= 0)
+	{
+		/* Ouch - an unnecessary lseek() each time through the loop! */
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+		/* Extend by one page. */
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+		page = BufferGetPage(buffer);
+		PageInit(page, BufferGetPageSize(buffer), 0);
+		MarkBufferDirty(buffer);
+		blockNum = BufferGetBlockNumber(buffer);
+		UnlockReleaseBuffer(buffer);
+
+		if (firstBlock == InvalidBlockNumber)
+			firstBlock = blockNum;
+
+		lastBlock = blockNum;
+	}
+
+	/*
+	 * Put the page in the freespace map so other backends can find it.
+	 * This is what will keep those other backends from also queueing up
+	 * on the relation extension lock.
+	 */
+	FreeSpaceMapBulkExtend(relation, firstBlock, lastBlock);
+}
+
+/*
  * RelationGetBufferForTuple
  *
  *	Returns pinned and exclusive-locked buffer of a page in given relation
@@ -233,8 +289,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	bool		use_fsm = !(options & HEAP_INSERT_SKIP_FSM);
 	Buffer		buffer = InvalidBuffer;
 	Page		page;
-	Size		pageFreeSpace,
-				saveFreeSpace;
+	Size		pageFreeSpace = 0,
+				saveFreeSpace = 0;
 	BlockNumber targetBlock,
 				otherBlock;
 	bool		needLock;
@@ -308,6 +364,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		}
 	}
 
+loop:
 	while (targetBlock != InvalidBlockNumber)
 	{
 		/*
@@ -440,10 +497,54 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	 */
 	needLock = !RELATION_IS_LOCAL(relation);
 
+	/*
+	 * If we need the lock but are not able to acquire it immediately, we'll
+	 * consider extending the relation by multiple blocks at a time to manage
+	 * contention on the relation extension lock.  However, this only makes
+	 * sense if we're using the FSM; otherwise, there's no point.
+	 */
 	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
+	{
+		if (!use_fsm)
+			LockRelationForExtension(relation, ExclusiveLock);
+		else if (!ConditionLockRelationForExtension(relation, ExclusiveLock))
+		{
+			BlockNumber last_blkno = RelationGetNumberOfBlocks(relation);
+
+			targetBlock = GetPageWithFreeSpaceExtended(relation,
+													last_blkno,
+													len + saveFreeSpace);
+			if (targetBlock != InvalidBlockNumber)
+				goto loop;
+
+			/* Couldn't get the lock immediately; wait for it. */
+			LockRelationForExtension(relation, ExclusiveLock);
+
+			/*
+			 * Check if some other backend has extended a block for us while
+			 * we were waiting on the lock.
+			 */
+			targetBlock = GetPageWithFreeSpace(relation, len + saveFreeSpace);
+
+			/*
+			 * If some other waiter has already extended the relation, we
+			 * don't need to do so; just use the existing freespace.
+			 */
+			if (targetBlock != InvalidBlockNumber)
+			{
+				UnlockRelationForExtension(relation, ExclusiveLock);
+				goto loop;
+			}
+
+			/* Time to bulk-extend. */
+			RelationAddExtraBlocks(relation, bistate);
+		}
+	}
 
 	/*
+	 * In addition to whatever extension we performed above, we always add
+	 * at least one block to satisfy our own request.
+	 *
 	 * XXX This does an lseek - rather expensive - but at the moment it is the
 	 * only way to accurately determine how many blocks are in a relation.  Is
 	 * it worth keeping an accurate file length in shared memory someplace,
diff --git a/src/backend/storage/freespace/freespace.c b/src/backend/storage/freespace/freespace.c
index 2631080..b3ce12e 100644
--- a/src/backend/storage/freespace/freespace.c
+++ b/src/backend/storage/freespace/freespace.c
@@ -109,8 +109,11 @@ static int fsm_set_and_search(Relation rel, FSMAddress addr, uint16 slot,
 				   uint8 newValue, uint8 minValue);
 static BlockNumber fsm_search(Relation rel, uint8 min_cat);
 static uint8 fsm_vacuum_page(Relation rel, FSMAddress addr, bool *eof);
-
-
+static bool fsm_addr_on_same_page(FSMAddress addr1, FSMAddress addr2);
+static void fsm_set_range(Relation rel, FSMAddress addr, uint16 firstSlot,
+					uint16 lastSlot, uint8 newValue);
+static void fsm_update_recursive(Relation rel, FSMAddress addr, uint8 new_cat);
+static int fsm_search_from_addr(Relation rel, FSMAddress addr, uint8 minValue);
 /******** Public API ********/
 
 /*
@@ -129,9 +132,46 @@ static uint8 fsm_vacuum_page(Relation rel, FSMAddress addr, bool *eof);
 BlockNumber
 GetPageWithFreeSpace(Relation rel, Size spaceNeeded)
 {
-	uint8		min_cat = fsm_space_needed_to_cat(spaceNeeded);
+	/*
+	 * Call GetPageWithFreeSpaceExtended with InvalidBlockNumber so that
+	 * it will search the FSM tree from the root
+	 */
+	return GetPageWithFreeSpaceExtended(rel, InvalidBlockNumber, spaceNeeded);
+}
 
-	return fsm_search(rel, min_cat);
+/*
+ * 		GetPageWithFreeSpaceExtended
+ *
+ * As above, but start the search from oldPage instead of staring from root
+ * So that, we can find the appropriate page in cases where free block is
+ * added to FSM but not yet updated up till root. If oldpage is Invalid
+ * then start the search from root.
+ */
+BlockNumber
+GetPageWithFreeSpaceExtended(Relation rel, BlockNumber oldPage,
+								Size spaceNeeded)
+{
+	int			search_cat = fsm_space_needed_to_cat(spaceNeeded);
+	FSMAddress	addr;
+	uint16		slot;
+	int			search_slot = -1;
+
+	if (oldPage != InvalidBlockNumber)
+	{
+		/* Get the location of the FSM byte representing the heap block */
+		addr = fsm_get_location(oldPage, &slot);
+
+		search_slot = fsm_search_from_addr(rel, addr, search_cat);
+	}
+
+	/*
+	 * If fsm_search_from_addr found a suitable new block, return that.
+	 * Otherwise, search as usual.
+	 */
+	if (search_slot != -1)
+		return fsm_get_heap_blk(addr, search_slot);
+	else
+		return fsm_search(rel, search_cat);
 }
 
 /*
@@ -189,6 +229,55 @@ RecordPageWithFreeSpace(Relation rel, BlockNumber heapBlk, Size spaceAvail)
 }
 
 /*
+ * Sets all the FSM entries for pages between firstBlock and lastBlock to
+ * BLKSIZE and then bubbles that up to the higher levels of the tree and
+ * all the way to the root.
+ */
+void
+FreeSpaceMapBulkExtend(Relation rel, BlockNumber firstBlock,
+					BlockNumber lastBlock)
+{
+	int			new_cat = fsm_space_avail_to_cat(BLCKSZ);
+	FSMAddress	addr1,
+				addr2;
+	uint16		firstSlot;
+	uint16		lastSlot;
+	uint16		slot;
+	BlockNumber	blockNum,
+				startBlk;
+
+	startBlk = blockNum = firstBlock;
+	addr1 = fsm_get_location(firstBlock, &slot);
+
+	while (blockNum <= lastBlock)
+	{
+		addr2 = fsm_get_location(blockNum+1, &slot);
+
+		/*
+		 * If next block is on different FSM page or we reached to the last
+		 * block then add what we got so far
+		 */
+		if (!fsm_addr_on_same_page(addr1, addr2)
+			|| blockNum == lastBlock)
+		{
+			/* This BLock is on the next FSM page so update the FSM page */
+			fsm_get_location(startBlk, &firstSlot);
+			fsm_get_location(blockNum, &lastSlot);
+
+			fsm_set_range(rel, addr1, firstSlot, lastSlot, new_cat);
+			fsm_update_recursive(rel, addr1, new_cat);
+			/*
+			 * Continue updating FSM till last page, so set the addr1 as new
+			 * Address. and continue search for this page.
+			 */
+			addr1 = addr2;
+		}
+
+		blockNum++;
+	}
+}
+
+/*
  * XLogRecordPageWithFreeSpace - like RecordPageWithFreeSpace, for use in
  *		WAL replay
  */
@@ -788,3 +877,85 @@ fsm_vacuum_page(Relation rel, FSMAddress addr, bool *eof_p)
 
 	return max_avail;
 }
+
+static bool
+fsm_addr_on_same_page(FSMAddress addr1, FSMAddress addr2)
+{
+	Assert(addr1);
+	Assert(addr2);
+
+	if ((addr1.level != addr2.level)
+		|| (addr1.logpageno != addr2.logpageno))
+		return false;
+
+	return true;
+}
+
+/*
+ * Set value in given FSM page for given slot range.
+ */
+static void
+fsm_set_range(Relation rel, FSMAddress addr, uint16 firstSlot,
+			uint16 lastSlot, uint8 newValue)
+{
+	Buffer		buf;
+	Page		page;
+	int			slot;
+
+	buf = fsm_readbuf(rel, addr, true);
+	LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
+
+	page = BufferGetPage(buf);
+
+	slot = firstSlot;
+
+	for (slot = firstSlot; slot <= lastSlot; slot++)
+		fsm_set_avail(page, slot, newValue);
+
+	MarkBufferDirtyHint(buf, false);
+
+	UnlockReleaseBuffer(buf);
+}
+
+static void
+fsm_update_recursive(Relation rel, FSMAddress addr, uint8 new_cat)
+{
+	uint16		parentslot;
+	FSMAddress	parent;
+
+	if (addr.level == FSM_ROOT_LEVEL)
+		return;
+
+	parent = fsm_get_parent(addr, &parentslot);
+	fsm_set_and_search(rel, parent, parentslot, new_cat, 0);
+	fsm_update_recursive(rel, parent, new_cat);
+}
+
+/*
+ * Search the the fsm tree for for free space > minValue
+ * It will start the search from given addr, and will be used for searching
+ * the required page in cases where vacuum have not yet updated the FSM tree
+ * till root level.
+ * If one is found, its slot number is returned, -1 otherwise.
+ */
+static int
+fsm_search_from_addr(Relation rel, FSMAddress addr, uint8 minValue)
+{
+	Buffer		buf;
+	int			newslot = -1;
+
+	buf = fsm_readbuf(rel, addr, true);
+	LockBuffer(buf, BUFFER_LOCK_SHARE);
+
+	Assert(minValue != 0);
+
+	/* Search while we still hold the lock */
+	newslot = fsm_search_avail(buf, minValue,
+							addr.level == FSM_BOTTOM_LEVEL,
+							false);
+
+	UnlockReleaseBuffer(buf);
+
+	return newslot;
+}
+
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 0632fc0..7e04137 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -341,6 +341,41 @@ LockRelationForExtension(Relation relation, LOCKMODE lockmode)
 }
 
 /*
+ *		ConditionLockRelationForExtension
+ *
+ * As above, but only lock if we can get the lock without blocking.
+ * Returns TRUE iff the lock was acquired.
+ */
+bool
+ConditionLockRelationForExtension(Relation relation, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return (LockAcquire(&tag, lockmode, false, true) != LOCKACQUIRE_NOT_AVAIL);
+}
+
+/*
+ *		RelationExtensionLockWaiterCount
+ *
+ * Count the lock requester for the RelationExtension lock.
+ */
+int
+RelationExtensionLockWaiterCount(Relation relation)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockWaiterCount(&tag);
+}
+
+/*
  *		UnlockRelationForExtension
  */
 void
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index b30b7b1..353f705 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -4380,3 +4380,40 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	LockRelease(&tag, ShareLock, false);
 	return true;
 }
+
+/*
+ * LockWaiterCount
+ *
+ * Find the number of lock requester on this locktag
+ */
+int
+LockWaiterCount(const LOCKTAG *locktag)
+{
+	LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
+	LOCK	   *lock;
+	bool		found;
+	uint32		hashcode;
+	LWLock		*partitionLock;
+	int			waiters = 0;
+
+	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
+		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
+
+	hashcode = LockTagHashCode(locktag);
+	partitionLock = LockHashPartitionLock(hashcode);
+	LWLockAcquire(partitionLock, LW_EXCLUSIVE);
+
+	lock = (LOCK *) hash_search_with_hash_value(LockMethodLockHash,
+												(const void *) locktag,
+												hashcode,
+												HASH_FIND,
+												&found);
+	if (found)
+	{
+		Assert(lock != NULL);
+		waiters = lock->nRequested;
+	}
+	LWLockRelease(partitionLock);
+
+	return waiters;
+}
diff --git a/src/include/storage/freespace.h b/src/include/storage/freespace.h
index 19dcb8d..8da5312 100644
--- a/src/include/storage/freespace.h
+++ b/src/include/storage/freespace.h
@@ -32,5 +32,10 @@ extern void XLogRecordPageWithFreeSpace(RelFileNode rnode, BlockNumber heapBlk,
 
 extern void FreeSpaceMapTruncateRel(Relation rel, BlockNumber nblocks);
 extern void FreeSpaceMapVacuum(Relation rel);
+extern void FreeSpaceMapBulkExtend(Relation rel, BlockNumber firstBlock,
+							BlockNumber lastBlock);
+extern BlockNumber GetPageWithFreeSpaceExtended(Relation rel,
+								BlockNumber oldPage,
+								Size spaceNeeded);
 
 #endif   /* FREESPACE_H_ */
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index 975b6f8..4460756 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -53,6 +53,9 @@ extern void UnlockRelationIdForSession(LockRelId *relid, LOCKMODE lockmode);
 /* Lock a relation for extension */
 extern void LockRelationForExtension(Relation relation, LOCKMODE lockmode);
 extern void UnlockRelationForExtension(Relation relation, LOCKMODE lockmode);
+extern bool ConditionLockRelationForExtension(Relation relation,
+								  LOCKMODE lockmode);
+extern int	RelationExtensionLockWaiterCount(Relation relation);
 
 /* Lock a page (currently only used within indexes) */
 extern void LockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode);
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index b26427d..9c08679 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -574,6 +574,8 @@ extern void RememberSimpleDeadLock(PGPROC *proc1,
 					   PGPROC *proc2);
 extern void InitDeadLockChecking(void);
 
+extern int LockWaiterCount(const LOCKTAG *locktag);
+
 #ifdef LOCK_DEBUG
 extern void DumpLocks(PGPROC *proc);
 extern void DumpAllLocks(void);

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..7d322f8 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -169,6 +169,62 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 }
 
 /*
+ * Extend a relation by multiple blocks to avoid future contention on the
+ * relation extension lock.  Our goal is to pre-extend the relation by an
+ * amount which ramps up as the degree of contention ramps up, but limiting
+ * the result to some sane overall value.
+ */
+static void
+RelationAddExtraBlocks(Relation relation, BulkInsertState bistate)
+{
+	Page		page;
+	BlockNumber blockNum,
+				firstBlock,
+				lastBlock;
+	int			extraBlocks = 0;
+	int			lockWaiters = 0;
+	Buffer		buffer;
+
+	/*
+	 * We use the length of the lock wait queue to judge how much to extend.
+	 * It might seem like multiplying the number of lock waiters by as much
+	 * as 20 is too aggressive, but benchmarking revealed that smaller numbers
+	 * were insufficient.  512 is just an arbitrary cap to prevent pathological
+	 * results (and excessive wasted disk space).
+	 */
+	lockWaiters = RelationExtensionLockWaiterCount(relation);
+	extraBlocks = Min(512, lockWaiters * 20);
+
+	firstBlock = lastBlock = InvalidBlockNumber;
+
+	while (extraBlocks-- >= 0)
+	{
+		/* Ouch - an unnecessary lseek() each time through the loop! */
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+		/* Extend by one page. */
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+		page = BufferGetPage(buffer);
+		PageInit(page, BufferGetPageSize(buffer), 0);
+		MarkBufferDirty(buffer);
+		blockNum = BufferGetBlockNumber(buffer);
+		UnlockReleaseBuffer(buffer);
+
+		if (firstBlock == InvalidBlockNumber)
+			firstBlock = blockNum;
+
+		lastBlock = blockNum;
+	}
+
+	/*
+	 * Put the page in the freespace map so other backends can find it.
+	 * This is what will keep those other backends from also queueing up
+	 * on the relation extension lock.
+	 */
+	FreeSpaceMapBulkExtend(relation, firstBlock, lastBlock);
+}
+
+/*
  * RelationGetBufferForTuple
  *
  *	Returns pinned and exclusive-locked buffer of a page in given relation
@@ -233,8 +289,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	bool		use_fsm = !(options & HEAP_INSERT_SKIP_FSM);
 	Buffer		buffer = InvalidBuffer;
 	Page		page;
-	Size		pageFreeSpace,
-				saveFreeSpace;
+	Size		pageFreeSpace = 0,
+				saveFreeSpace = 0;
 	BlockNumber targetBlock,
 				otherBlock;
 	bool		needLock;
@@ -308,6 +364,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		}
 	}
 
+loop:
 	while (targetBlock != InvalidBlockNumber)
 	{
 		/*
@@ -440,10 +497,54 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	 */
 	needLock = !RELATION_IS_LOCAL(relation);
 
+	/*
+	 * If we need the lock but are not able to acquire it immediately, we'll
+	 * consider extending the relation by multiple blocks at a time to manage
+	 * contention on the relation extension lock.  However, this only makes
+	 * sense if we're using the FSM; otherwise, there's no point.
+	 */
 	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
+	{
+		if (!use_fsm)
+			LockRelationForExtension(relation, ExclusiveLock);
+		else if (!ConditionLockRelationForExtension(relation, ExclusiveLock))
+		{
+			BlockNumber last_blkno = RelationGetNumberOfBlocks(relation);
+
+			targetBlock = GetPageWithFreeSpaceExtended(relation,
+													last_blkno,
+													len + saveFreeSpace);
+			if (targetBlock != InvalidBlockNumber)
+				goto loop;
+
+			/* Couldn't get the lock immediately; wait for it. */
+			LockRelationForExtension(relation, ExclusiveLock);
+
+			/*
+			 * Check if some other backend has extended a block for us while
+			 * we were waiting on the lock.
+			 */
+			targetBlock = GetPageWithFreeSpace(relation, len + saveFreeSpace);
+
+			/*
+			 * If some other waiter has already extended the relation, we
+			 * don't need to do so; just use the existing freespace.
+			 */
+			if (targetBlock != InvalidBlockNumber)
+			{
+				UnlockRelationForExtension(relation, ExclusiveLock);
+				goto loop;
+			}
+
+			/* Time to bulk-extend. */
+			RelationAddExtraBlocks(relation, bistate);
+		}
+	}
 
 	/*
+	 * In addition to whatever extension we performed above, we always add
+	 * at least one block to satisfy our own request.
+	 *
 	 * XXX This does an lseek - rather expensive - but at the moment it is the
 	 * only way to accurately determine how many blocks are in a relation.  Is
 	 * it worth keeping an accurate file length in shared memory someplace,
diff --git a/src/backend/storage/freespace/freespace.c b/src/backend/storage/freespace/freespace.c
index 2631080..b3ce12e 100644
--- a/src/backend/storage/freespace/freespace.c
+++ b/src/backend/storage/freespace/freespace.c
@@ -109,8 +109,11 @@ static int fsm_set_and_search(Relation rel, FSMAddress addr, uint16 slot,
 				   uint8 newValue, uint8 minValue);
 static BlockNumber fsm_search(Relation rel, uint8 min_cat);
 static uint8 fsm_vacuum_page(Relation rel, FSMAddress addr, bool *eof);
-
-
+static bool fsm_addr_on_same_page(FSMAddress addr1, FSMAddress addr2);
+static void fsm_set_range(Relation rel, FSMAddress addr, uint16 firstSlot,
+					uint16 lastSlot, uint8 newValue);
+static void fsm_update_recursive(Relation rel, FSMAddress addr, uint8 new_cat);
+static int fsm_search_from_addr(Relation rel, FSMAddress addr, uint8 minValue);
 /******** Public API ********/
 
 /*
@@ -129,9 +132,46 @@ static uint8 fsm_vacuum_page(Relation rel, FSMAddress addr, bool *eof);
 BlockNumber
 GetPageWithFreeSpace(Relation rel, Size spaceNeeded)
 {
-	uint8		min_cat = fsm_space_needed_to_cat(spaceNeeded);
+	/*
+	 * Call GetPageWithFreeSpaceExtended with InvalidBlockNumber so that
+	 * it will search the FSM tree from the root
+	 */
+	return GetPageWithFreeSpaceExtended(rel, InvalidBlockNumber, spaceNeeded);
+}
 
-	return fsm_search(rel, min_cat);
+/*
+ * 		GetPageWithFreeSpaceExtended
+ *
+ * As above, but start the search from oldPage instead of staring from root
+ * So that, we can find the appropriate page in cases where free block is
+ * added to FSM but not yet updated up till root. If oldpage is Invalid
+ * then start the search from root.
+ */
+BlockNumber
+GetPageWithFreeSpaceExtended(Relation rel, BlockNumber oldPage,
+								Size spaceNeeded)
+{
+	int			search_cat = fsm_space_needed_to_cat(spaceNeeded);
+	FSMAddress	addr;
+	uint16		slot;
+	int			search_slot = -1;
+
+	if (oldPage != InvalidBlockNumber)
+	{
+		/* Get the location of the FSM byte representing the heap block */
+		addr = fsm_get_location(oldPage, &slot);
+
+		search_slot = fsm_search_from_addr(rel, addr, search_cat);
+	}
+
+	/*
+	 * If fsm_search_from_addr found a suitable new block, return that.
+	 * Otherwise, search as usual.
+	 */
+	if (search_slot != -1)
+		return fsm_get_heap_blk(addr, search_slot);
+	else
+		return fsm_search(rel, search_cat);
 }
 
 /*
@@ -189,6 +229,55 @@ RecordPageWithFreeSpace(Relation rel, BlockNumber heapBlk, Size spaceAvail)
 }
 
 /*
+ * Sets all the FSM entries for pages between firstBlock and lastBlock to
+ * BLKSIZE and then bubbles that up to the higher levels of the tree and
+ * all the way to the root.
+ */
+void
+FreeSpaceMapBulkExtend(Relation rel, BlockNumber firstBlock,
+					BlockNumber lastBlock)
+{
+	int			new_cat = fsm_space_avail_to_cat(BLCKSZ);
+	FSMAddress	addr1,
+				addr2;
+	uint16		firstSlot;
+	uint16		lastSlot;
+	uint16		slot;
+	BlockNumber	blockNum,
+				startBlk;
+
+	startBlk = blockNum = firstBlock;
+	addr1 = fsm_get_location(firstBlock, &slot);
+
+	while (blockNum <= lastBlock)
+	{
+		addr2 = fsm_get_location(blockNum+1, &slot);
+
+		/*
+		 * If next block is on different FSM page or we reached to the last
+		 * block then add what we got so far
+		 */
+		if (!fsm_addr_on_same_page(addr1, addr2)
+			|| blockNum == lastBlock)
+		{
+			/* This BLock is on the next FSM page so update the FSM page */
+			fsm_get_location(startBlk, &firstSlot);
+			fsm_get_location(blockNum, &lastSlot);
+
+			fsm_set_range(rel, addr1, firstSlot, lastSlot, new_cat);
+			fsm_update_recursive(rel, addr1, new_cat);
+			/*
+			 * Continue updating FSM till last page, so set the addr1 as new
+			 * Address. and continue search for this page.
+			 */
+			addr1 = addr2;
+		}
+
+		blockNum++;
+	}
+}
+
+/*
  * XLogRecordPageWithFreeSpace - like RecordPageWithFreeSpace, for use in
  *		WAL replay
  */
@@ -788,3 +877,85 @@ fsm_vacuum_page(Relation rel, FSMAddress addr, bool *eof_p)
 
 	return max_avail;
 }
+
+static bool
+fsm_addr_on_same_page(FSMAddress addr1, FSMAddress addr2)
+{
+	Assert(addr1);
+	Assert(addr2);
+
+	if ((addr1.level != addr2.level)
+		|| (addr1.logpageno != addr2.logpageno))
+		return false;
+
+	return true;
+}
+
+/*
+ * Set value in given FSM page for given slot range.
+ */
+static void
+fsm_set_range(Relation rel, FSMAddress addr, uint16 firstSlot,
+			uint16 lastSlot, uint8 newValue)
+{
+	Buffer		buf;
+	Page		page;
+	int			slot;
+
+	buf = fsm_readbuf(rel, addr, true);
+	LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
+
+	page = BufferGetPage(buf);
+
+	slot = firstSlot;
+
+	for (slot = firstSlot; slot <= lastSlot; slot++)
+		fsm_set_avail(page, slot, newValue);
+
+	MarkBufferDirtyHint(buf, false);
+
+	UnlockReleaseBuffer(buf);
+}
+
+static void
+fsm_update_recursive(Relation rel, FSMAddress addr, uint8 new_cat)
+{
+	uint16		parentslot;
+	FSMAddress	parent;
+
+	if (addr.level == FSM_ROOT_LEVEL)
+		return;
+
+	parent = fsm_get_parent(addr, &parentslot);
+	fsm_set_and_search(rel, parent, parentslot, new_cat, 0);
+	fsm_update_recursive(rel, parent, new_cat);
+}
+
+/*
+ * Search the the fsm tree for for free space > minValue
+ * It will start the search from given addr, and will be used for searching
+ * the required page in cases where vacuum have not yet updated the FSM tree
+ * till root level.
+ * If one is found, its slot number is returned, -1 otherwise.
+ */
+static int
+fsm_search_from_addr(Relation rel, FSMAddress addr, uint8 minValue)
+{
+	Buffer		buf;
+	int			newslot = -1;
+
+	buf = fsm_readbuf(rel, addr, true);
+	LockBuffer(buf, BUFFER_LOCK_SHARE);
+
+	Assert(minValue != 0);
+
+	/* Search while we still hold the lock */
+	newslot = fsm_search_avail(buf, minValue,
+							addr.level == FSM_BOTTOM_LEVEL,
+							false);
+
+	UnlockReleaseBuffer(buf);
+
+	return newslot;
+}
+
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 0632fc0..7e04137 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -341,6 +341,41 @@ LockRelationForExtension(Relation relation, LOCKMODE lockmode)
 }
 
 /*
+ *		ConditionLockRelationForExtension
+ *
+ * As above, but only lock if we can get the lock without blocking.
+ * Returns TRUE iff the lock was acquired.
+ */
+bool
+ConditionLockRelationForExtension(Relation relation, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return (LockAcquire(&tag, lockmode, false, true) != LOCKACQUIRE_NOT_AVAIL);
+}
+
+/*
+ *		RelationExtensionLockWaiterCount
+ *
+ * Count the lock requester for the RelationExtension lock.
+ */
+int
+RelationExtensionLockWaiterCount(Relation relation)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockWaiterCount(&tag);
+}
+
+/*
  *		UnlockRelationForExtension
  */
 void
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index b30b7b1..353f705 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -4380,3 +4380,40 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	LockRelease(&tag, ShareLock, false);
 	return true;
 }
+
+/*
+ * LockWaiterCount
+ *
+ * Find the number of lock requester on this locktag
+ */
+int
+LockWaiterCount(const LOCKTAG *locktag)
+{
+	LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
+	LOCK	   *lock;
+	bool		found;
+	uint32		hashcode;
+	LWLock		*partitionLock;
+	int			waiters = 0;
+
+	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
+		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
+
+	hashcode = LockTagHashCode(locktag);
+	partitionLock = LockHashPartitionLock(hashcode);
+	LWLockAcquire(partitionLock, LW_EXCLUSIVE);
+
+	lock = (LOCK *) hash_search_with_hash_value(LockMethodLockHash,
+												(const void *) locktag,
+												hashcode,
+												HASH_FIND,
+												&found);
+	if (found)
+	{
+		Assert(lock != NULL);
+		waiters = lock->nRequested;
+	}
+	LWLockRelease(partitionLock);
+
+	return waiters;
+}
diff --git a/src/include/storage/freespace.h b/src/include/storage/freespace.h
index 19dcb8d..8da5312 100644
--- a/src/include/storage/freespace.h
+++ b/src/include/storage/freespace.h
@@ -32,5 +32,10 @@ extern void XLogRecordPageWithFreeSpace(RelFileNode rnode, BlockNumber heapBlk,
 
 extern void FreeSpaceMapTruncateRel(Relation rel, BlockNumber nblocks);
 extern void FreeSpaceMapVacuum(Relation rel);
+extern void FreeSpaceMapBulkExtend(Relation rel, BlockNumber firstBlock,
+							BlockNumber lastBlock);
+extern BlockNumber GetPageWithFreeSpaceExtended(Relation rel,
+								BlockNumber oldPage,
+								Size spaceNeeded);
 
 #endif   /* FREESPACE_H_ */
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index 975b6f8..4460756 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -53,6 +53,9 @@ extern void UnlockRelationIdForSession(LockRelId *relid, LOCKMODE lockmode);
 /* Lock a relation for extension */
 extern void LockRelationForExtension(Relation relation, LOCKMODE lockmode);
 extern void UnlockRelationForExtension(Relation relation, LOCKMODE lockmode);
+extern bool ConditionLockRelationForExtension(Relation relation,
+								  LOCKMODE lockmode);
+extern int	RelationExtensionLockWaiterCount(Relation relation);
 
 /* Lock a page (currently only used within indexes) */
 extern void LockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode);
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index b26427d..9c08679 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -574,6 +574,8 @@ extern void RememberSimpleDeadLock(PGPROC *proc1,
 					   PGPROC *proc2);
 extern void InitDeadLockChecking(void);
 
+extern int LockWaiterCount(const LOCKTAG *locktag);
+
 #ifdef LOCK_DEBUG
 extern void DumpLocks(PGPROC *proc);
 extern void DumpAllLocks(void);

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..54ffcd4 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -169,6 +169,55 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 }
 
 /*
+ * Extend a relation by multiple blocks to avoid future contention on the
+ * relation extension lock.  Our goal is to pre-extend the relation by an
+ * amount which ramps up as the degree of contention ramps up, but limiting
+ * the result to some sane overall value.
+ */
+static void
+RelationAddExtraBlocks(Relation relation, BulkInsertState bistate)
+{
+	Page		page;
+	Size		freespace;
+	BlockNumber blockNum;
+	int			extraBlocks = 0;
+	int			lockWaiters = 0;
+	Buffer		buffer;
+
+	/*
+	 * We use the length of the lock wait queue to judge how much to extend.
+	 * It might seem like multiplying the number of lock waiters by as much
+	 * as 20 is too aggressive, but benchmarking revealed that smaller numbers
+	 * were insufficient.  512 is just an arbitrary cap to prevent pathological
+	 * results (and excessive wasted disk space).
+	 */
+	lockWaiters = RelationExtensionLockWaiterCount(relation);
+	extraBlocks = Min(512, lockWaiters * 20);
+
+	while (extraBlocks-- >= 0)
+	{
+		/* Ouch - an unnecessary lseek() each time through the loop! */
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+		/* Extend by one page. */
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+		page = BufferGetPage(buffer);
+		PageInit(page, BufferGetPageSize(buffer), 0);
+		freespace = PageGetHeapFreeSpace(page);
+		MarkBufferDirty(buffer);
+		blockNum = BufferGetBlockNumber(buffer);
+		UnlockReleaseBuffer(buffer);
+
+		/*
+		 * Put the page in the freespace map so other backends can find it.
+		 * This is what will keep those other backends from also queueing up
+		 * on the relation extension lock.
+		 */
+		RecordPageWithFreeSpace(relation, blockNum, freespace);
+	}
+}
+
+/*
  * RelationGetBufferForTuple
  *
  *	Returns pinned and exclusive-locked buffer of a page in given relation
@@ -233,10 +282,11 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	bool		use_fsm = !(options & HEAP_INSERT_SKIP_FSM);
 	Buffer		buffer = InvalidBuffer;
 	Page		page;
-	Size		pageFreeSpace,
-				saveFreeSpace;
+	Size		pageFreeSpace = 0,
+				saveFreeSpace = 0;
 	BlockNumber targetBlock,
-				otherBlock;
+				otherBlock,
+				lastValidBlock = InvalidBlockNumber;
 	bool		needLock;
 
 	len = MAXALIGN(len);		/* be conservative */
@@ -308,6 +358,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		}
 	}
 
+loop:
 	while (targetBlock != InvalidBlockNumber)
 	{
 		/*
@@ -388,6 +439,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 								 otherBlock, targetBlock, vmbuffer_other,
 								 vmbuffer);
 
+		lastValidBlock = targetBlock;
+
 		/*
 		 * Now we can check to see if there's enough free space here. If so,
 		 * we're done.
@@ -440,10 +493,68 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	 */
 	needLock = !RELATION_IS_LOCAL(relation);
 
+	/*
+	 * If we need the lock but are not able to acquire it immediately, we'll
+	 * consider extending the relation by multiple blocks at a time to manage
+	 * contention on the relation extension lock.  However, this only makes
+	 * sense if we're using the FSM; otherwise, there's no point.
+	 */
 	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
+	{
+		if (!use_fsm)
+			LockRelationForExtension(relation, ExclusiveLock);
+		else if (!ConditionLockRelationForExtension(relation, ExclusiveLock))
+		{
+			BlockNumber     last_blkno;
+			BlockNumber     prev_blkno;
+
+			/* Couldn't get the lock immmediately; wait for it. */
+			LockRelationForExtension(relation, ExclusiveLock);
+
+			last_blkno = RelationGetNumberOfBlocks(relation);
+
+			/*
+			 * If lastValidBlock is Invalid, it means when we checked there
+			 * was no block in the relation, but still we need to search
+			 * because other backend might have extended it
+			 */
+			if (lastValidBlock != InvalidBlockNumber)
+				prev_blkno = lastValidBlock;
+			else
+				prev_blkno = last_blkno;
+
+			/*
+			 * Here we are calling GetPageWithFreeSpaceExtended with
+			 * prev block and the last block of the relation. So that
+			 * we can find any block added in FSM near to the prev valid
+			 * block and also if bulk extend has spill over to next FSM
+			 * page, one bulk extend max limit is 512 pages, so finding
+			 * in two FSM page is enough.
+			 */
+			targetBlock = GetPageWithFreeSpaceExtended(relation,
+													prev_blkno,
+													last_blkno,
+													len + saveFreeSpace);
+
+			/*
+			 * If some other waiter has already extended the relation, we
+			 * don't need to do so; just use the existing freespace.
+			 */
+			if (targetBlock != InvalidBlockNumber)
+			{
+				UnlockRelationForExtension(relation, ExclusiveLock);
+				goto loop;
+			}
+
+			/* Time to bulk-extend. */
+			RelationAddExtraBlocks(relation, bistate);
+		}
+	}
 
 	/*
+	 * In addition to whatever extension we performed above, we always add
+	 * at least one block to satisfy our own request.
+	 *
 	 * XXX This does an lseek - rather expensive - but at the moment it is the
 	 * only way to accurately determine how many blocks are in a relation.  Is
 	 * it worth keeping an accurate file length in shared memory someplace,
diff --git a/src/backend/storage/freespace/freespace.c b/src/backend/storage/freespace/freespace.c
index 2631080..19e499b 100644
--- a/src/backend/storage/freespace/freespace.c
+++ b/src/backend/storage/freespace/freespace.c
@@ -109,7 +109,8 @@ static int fsm_set_and_search(Relation rel, FSMAddress addr, uint16 slot,
 				   uint8 newValue, uint8 minValue);
 static BlockNumber fsm_search(Relation rel, uint8 min_cat);
 static uint8 fsm_vacuum_page(Relation rel, FSMAddress addr, bool *eof);
-
+static int fsm_search_from_addr(Relation rel, FSMAddress addr, uint8 minValue);
+static bool fsm_addr_on_same_page(FSMAddress addr1, FSMAddress addr2);
 
 /******** Public API ********/
 
@@ -129,9 +130,56 @@ static uint8 fsm_vacuum_page(Relation rel, FSMAddress addr, bool *eof);
 BlockNumber
 GetPageWithFreeSpace(Relation rel, Size spaceNeeded)
 {
-	uint8		min_cat = fsm_space_needed_to_cat(spaceNeeded);
+	/*
+	 * Call GetPageWithFreeSpaceExtended with InvalidBlockNumber so that
+	 * it will search the FSM tree from the root
+	 */
+	return GetPageWithFreeSpaceExtended(rel, InvalidBlockNumber, InvalidBlockNumber, spaceNeeded);
+}
+
+/*
+ * 		GetPageWithFreeSpaceExtended
+ *
+ * As above, but start the search in FSM page, where oldBlockNum has its slot
+ * instead of staring from root So that, we can find the appropriate page in
+ * cases where free block is added to FSM but not yet updated up till root.
+ * If oldBlockNum is Invalid then start the search from root. Also check in
+ * newBlockNum is in other FSM page, then search in that FSM page also.
+ */
+BlockNumber
+GetPageWithFreeSpaceExtended(Relation rel, BlockNumber oldBlockNum,
+							BlockNumber newBlockNum, Size spaceNeeded)
+{
+	int			search_cat = fsm_space_needed_to_cat(spaceNeeded);
+	FSMAddress	addr, new_addr;
+	uint16		slot;
+	int			search_slot = -1;
 
-	return fsm_search(rel, min_cat);
+	if (oldBlockNum != InvalidBlockNumber)
+	{
+		/* Get the location of the FSM byte representing the heap block */
+		addr = fsm_get_location(oldBlockNum, &slot);
+
+		search_slot = fsm_search_from_addr(rel, addr, search_cat);
+		if (search_slot == -1)
+		{
+			new_addr = fsm_get_location(newBlockNum, &slot);
+			if (!fsm_addr_on_same_page(addr, new_addr))
+			{
+				search_slot = fsm_search_from_addr(rel, new_addr, search_cat);
+				addr = new_addr;
+			}
+		}
+	}
+
+	/*
+	 * If fsm_search_from_addr found a suitable new block, return that.
+	 * Otherwise, search as usual.
+	 */
+	if (search_slot != -1)
+		return fsm_get_heap_blk(addr, search_slot);
+	else
+		return fsm_search(rel, search_cat);
 }
 
 /*
@@ -634,6 +682,44 @@ fsm_set_and_search(Relation rel, FSMAddress addr, uint16 slot,
 }
 
 /*
+ * Search the the fsm tree for for free space > minValue
+ * It will start the search from given addr, and will be used for searching
+ * the required page in cases where vacuum have not yet updated the FSM tree
+ * till root level.
+ * If one is found, its slot number is returned, -1 otherwise.
+ */
+static int
+fsm_search_from_addr(Relation rel, FSMAddress addr, uint8 minValue)
+{
+	Buffer		buf;
+	int			newslot = -1;
+
+	buf = fsm_readbuf(rel, addr, true);
+	LockBuffer(buf, BUFFER_LOCK_SHARE);
+
+	Assert(minValue != 0);
+
+	/* Search while we still hold the lock */
+	newslot = fsm_search_avail(buf, minValue,
+							addr.level == FSM_BOTTOM_LEVEL,
+							false);
+
+	UnlockReleaseBuffer(buf);
+
+	return newslot;
+}
+
+static bool
+fsm_addr_on_same_page(FSMAddress addr1, FSMAddress addr2)
+{
+	if ((addr1.level != addr2.level)
+		|| (addr1.logpageno != addr2.logpageno))
+		return false;
+
+	return true;
+}
+
+/*
  * Search the tree for a heap page with at least min_cat of free space
  */
 static BlockNumber
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 0632fc0..7e04137 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -341,6 +341,41 @@ LockRelationForExtension(Relation relation, LOCKMODE lockmode)
 }
 
 /*
+ *		ConditionLockRelationForExtension
+ *
+ * As above, but only lock if we can get the lock without blocking.
+ * Returns TRUE iff the lock was acquired.
+ */
+bool
+ConditionLockRelationForExtension(Relation relation, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return (LockAcquire(&tag, lockmode, false, true) != LOCKACQUIRE_NOT_AVAIL);
+}
+
+/*
+ *		RelationExtensionLockWaiterCount
+ *
+ * Count the lock requester for the RelationExtension lock.
+ */
+int
+RelationExtensionLockWaiterCount(Relation relation)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockWaiterCount(&tag);
+}
+
+/*
  *		UnlockRelationForExtension
  */
 void
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index b30b7b1..353f705 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -4380,3 +4380,40 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	LockRelease(&tag, ShareLock, false);
 	return true;
 }
+
+/*
+ * LockWaiterCount
+ *
+ * Find the number of lock requester on this locktag
+ */
+int
+LockWaiterCount(const LOCKTAG *locktag)
+{
+	LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
+	LOCK	   *lock;
+	bool		found;
+	uint32		hashcode;
+	LWLock		*partitionLock;
+	int			waiters = 0;
+
+	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
+		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
+
+	hashcode = LockTagHashCode(locktag);
+	partitionLock = LockHashPartitionLock(hashcode);
+	LWLockAcquire(partitionLock, LW_EXCLUSIVE);
+
+	lock = (LOCK *) hash_search_with_hash_value(LockMethodLockHash,
+												(const void *) locktag,
+												hashcode,
+												HASH_FIND,
+												&found);
+	if (found)
+	{
+		Assert(lock != NULL);
+		waiters = lock->nRequested;
+	}
+	LWLockRelease(partitionLock);
+
+	return waiters;
+}
diff --git a/src/include/storage/freespace.h b/src/include/storage/freespace.h
index 19dcb8d..281137a 100644
--- a/src/include/storage/freespace.h
+++ b/src/include/storage/freespace.h
@@ -32,5 +32,10 @@ extern void XLogRecordPageWithFreeSpace(RelFileNode rnode, BlockNumber heapBlk,
 
 extern void FreeSpaceMapTruncateRel(Relation rel, BlockNumber nblocks);
 extern void FreeSpaceMapVacuum(Relation rel);
+extern BlockNumber GetPageWithFreeSpaceExtended(Relation rel,
+								BlockNumber oldPage,
+								BlockNumber newPage,
+								Size spaceNeeded);
+
 
 #endif   /* FREESPACE_H_ */
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index 975b6f8..4460756 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -53,6 +53,9 @@ extern void UnlockRelationIdForSession(LockRelId *relid, LOCKMODE lockmode);
 /* Lock a relation for extension */
 extern void LockRelationForExtension(Relation relation, LOCKMODE lockmode);
 extern void UnlockRelationForExtension(Relation relation, LOCKMODE lockmode);
+extern bool ConditionLockRelationForExtension(Relation relation,
+								  LOCKMODE lockmode);
+extern int	RelationExtensionLockWaiterCount(Relation relation);
 
 /* Lock a page (currently only used within indexes) */
 extern void LockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode);
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index b26427d..9c08679 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -574,6 +574,8 @@ extern void RememberSimpleDeadLock(PGPROC *proc1,
 					   PGPROC *proc2);
 extern void InitDeadLockChecking(void);
 
+extern int LockWaiterCount(const LOCKTAG *locktag);
+
 #ifdef LOCK_DEBUG
 extern void DumpLocks(PGPROC *proc);
 extern void DumpAllLocks(void);

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 8140418..517d465 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -169,6 +169,55 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 }
 
 /*
+ * Extend a relation by multiple blocks to avoid future contention on the
+ * relation extension lock.  Our goal is to pre-extend the relation by an
+ * amount which ramps up as the degree of contention ramps up, but limiting
+ * the result to some sane overall value.
+ */
+static void
+RelationAddExtraBlocks(Relation relation, BulkInsertState bistate)
+{
+	Page		page;
+	Size		freespace;
+	BlockNumber blockNum;
+	int			extraBlocks = 0;
+	int			lockWaiters = 0;
+	Buffer		buffer;
+
+	/*
+	 * We use the length of the lock wait queue to judge how much to extend.
+	 * It might seem like multiplying the number of lock waiters by as much
+	 * as 20 is too aggressive, but benchmarking revealed that smaller numbers
+	 * were insufficient.  512 is just an arbitrary cap to prevent pathological
+	 * results (and excessive wasted disk space).
+	 */
+	lockWaiters = RelationExtensionLockWaiterCount(relation);
+	extraBlocks = Min(512, lockWaiters * 20);
+
+	while (extraBlocks-- >= 0)
+	{
+		/* Ouch - an unnecessary lseek() each time through the loop! */
+		buffer = ReadBufferBI(relation, P_NEW, bistate);
+
+		/* Extend by one page. */
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+		page = BufferGetPage(buffer);
+		PageInit(page, BufferGetPageSize(buffer), 0);
+		freespace = PageGetHeapFreeSpace(page);
+		MarkBufferDirty(buffer);
+		blockNum = BufferGetBlockNumber(buffer);
+		UnlockReleaseBuffer(buffer);
+
+		/*
+		 * Put the page in the freespace map so other backends can find it.
+		 * This is what will keep those other backends from also queueing up
+		 * on the relation extension lock.
+		 */
+		RecordPageWithFreeSpace(relation, blockNum, freespace);
+	}
+}
+
+/*
  * RelationGetBufferForTuple
  *
  *	Returns pinned and exclusive-locked buffer of a page in given relation
@@ -233,10 +282,11 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	bool		use_fsm = !(options & HEAP_INSERT_SKIP_FSM);
 	Buffer		buffer = InvalidBuffer;
 	Page		page;
-	Size		pageFreeSpace,
-				saveFreeSpace;
+	Size		pageFreeSpace = 0,
+				saveFreeSpace = 0;
 	BlockNumber targetBlock,
-				otherBlock;
+				otherBlock,
+				prev_blkno = RelationGetNumberOfBlocks(relation);
 	bool		needLock;
 
 	len = MAXALIGN(len);		/* be conservative */
@@ -308,6 +358,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		}
 	}
 
+loop:
 	while (targetBlock != InvalidBlockNumber)
 	{
 		/*
@@ -440,10 +491,65 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	 */
 	needLock = !RELATION_IS_LOCAL(relation);
 
+	/*
+	 * If we need the lock but are not able to acquire it immediately, we'll
+	 * consider extending the relation by multiple blocks at a time to manage
+	 * contention on the relation extension lock.  However, this only makes
+	 * sense if we're using the FSM; otherwise, there's no point.
+	 */
 	if (needLock)
-		LockRelationForExtension(relation, ExclusiveLock);
+	{
+		if (!use_fsm)
+			LockRelationForExtension(relation, ExclusiveLock);
+		else if (!ConditionLockRelationForExtension(relation, ExclusiveLock))
+		{
+			BlockNumber     last_blkno;
+
+			/* Couldn't get the lock immmediately; wait for it. */
+			LockRelationForExtension(relation, ExclusiveLock);
+
+			last_blkno = RelationGetNumberOfBlocks(relation);
+
+			/*
+			 * If lastValidBlock is Invalid, it means when we checked there
+			 * was no block in the relation, but still we need to search
+			 * because other backend might have extended it
+			 */
+			if (prev_blkno == InvalidBlockNumber)
+				prev_blkno = last_blkno;
+
+			/*
+			 * Here we are calling GetPageWithFreeSpaceExtended with
+			 * prev block and the last block of the relation. So that
+			 * we can find any block added in FSM near to the prev valid
+			 * block and also if bulk extend has spill over to next FSM
+			 * page, one bulk extend max limit is 512 pages, so finding
+			 * in two FSM page is enough.
+			 */
+			targetBlock = GetPageWithFreeSpaceExtended(relation,
+													prev_blkno,
+													last_blkno,
+													len + saveFreeSpace);
+
+			/*
+			 * If some other waiter has already extended the relation, we
+			 * don't need to do so; just use the existing freespace.
+			 */
+			if (targetBlock != InvalidBlockNumber)
+			{
+				UnlockRelationForExtension(relation, ExclusiveLock);
+				goto loop;
+			}
+
+			/* Time to bulk-extend. */
+			RelationAddExtraBlocks(relation, bistate);
+		}
+	}
 
 	/*
+	 * In addition to whatever extension we performed above, we always add
+	 * at least one block to satisfy our own request.
+	 *
 	 * XXX This does an lseek - rather expensive - but at the moment it is the
 	 * only way to accurately determine how many blocks are in a relation.  Is
 	 * it worth keeping an accurate file length in shared memory someplace,
diff --git a/src/backend/storage/freespace/freespace.c b/src/backend/storage/freespace/freespace.c
index 2631080..19e499b 100644
--- a/src/backend/storage/freespace/freespace.c
+++ b/src/backend/storage/freespace/freespace.c
@@ -109,7 +109,8 @@ static int fsm_set_and_search(Relation rel, FSMAddress addr, uint16 slot,
 				   uint8 newValue, uint8 minValue);
 static BlockNumber fsm_search(Relation rel, uint8 min_cat);
 static uint8 fsm_vacuum_page(Relation rel, FSMAddress addr, bool *eof);
-
+static int fsm_search_from_addr(Relation rel, FSMAddress addr, uint8 minValue);
+static bool fsm_addr_on_same_page(FSMAddress addr1, FSMAddress addr2);
 
 /******** Public API ********/
 
@@ -129,9 +130,56 @@ static uint8 fsm_vacuum_page(Relation rel, FSMAddress addr, bool *eof);
 BlockNumber
 GetPageWithFreeSpace(Relation rel, Size spaceNeeded)
 {
-	uint8		min_cat = fsm_space_needed_to_cat(spaceNeeded);
+	/*
+	 * Call GetPageWithFreeSpaceExtended with InvalidBlockNumber so that
+	 * it will search the FSM tree from the root
+	 */
+	return GetPageWithFreeSpaceExtended(rel, InvalidBlockNumber, InvalidBlockNumber, spaceNeeded);
+}
+
+/*
+ * 		GetPageWithFreeSpaceExtended
+ *
+ * As above, but start the search in FSM page, where oldBlockNum has its slot
+ * instead of staring from root So that, we can find the appropriate page in
+ * cases where free block is added to FSM but not yet updated up till root.
+ * If oldBlockNum is Invalid then start the search from root. Also check in
+ * newBlockNum is in other FSM page, then search in that FSM page also.
+ */
+BlockNumber
+GetPageWithFreeSpaceExtended(Relation rel, BlockNumber oldBlockNum,
+							BlockNumber newBlockNum, Size spaceNeeded)
+{
+	int			search_cat = fsm_space_needed_to_cat(spaceNeeded);
+	FSMAddress	addr, new_addr;
+	uint16		slot;
+	int			search_slot = -1;
 
-	return fsm_search(rel, min_cat);
+	if (oldBlockNum != InvalidBlockNumber)
+	{
+		/* Get the location of the FSM byte representing the heap block */
+		addr = fsm_get_location(oldBlockNum, &slot);
+
+		search_slot = fsm_search_from_addr(rel, addr, search_cat);
+		if (search_slot == -1)
+		{
+			new_addr = fsm_get_location(newBlockNum, &slot);
+			if (!fsm_addr_on_same_page(addr, new_addr))
+			{
+				search_slot = fsm_search_from_addr(rel, new_addr, search_cat);
+				addr = new_addr;
+			}
+		}
+	}
+
+	/*
+	 * If fsm_search_from_addr found a suitable new block, return that.
+	 * Otherwise, search as usual.
+	 */
+	if (search_slot != -1)
+		return fsm_get_heap_blk(addr, search_slot);
+	else
+		return fsm_search(rel, search_cat);
 }
 
 /*
@@ -634,6 +682,44 @@ fsm_set_and_search(Relation rel, FSMAddress addr, uint16 slot,
 }
 
 /*
+ * Search the the fsm tree for for free space > minValue
+ * It will start the search from given addr, and will be used for searching
+ * the required page in cases where vacuum have not yet updated the FSM tree
+ * till root level.
+ * If one is found, its slot number is returned, -1 otherwise.
+ */
+static int
+fsm_search_from_addr(Relation rel, FSMAddress addr, uint8 minValue)
+{
+	Buffer		buf;
+	int			newslot = -1;
+
+	buf = fsm_readbuf(rel, addr, true);
+	LockBuffer(buf, BUFFER_LOCK_SHARE);
+
+	Assert(minValue != 0);
+
+	/* Search while we still hold the lock */
+	newslot = fsm_search_avail(buf, minValue,
+							addr.level == FSM_BOTTOM_LEVEL,
+							false);
+
+	UnlockReleaseBuffer(buf);
+
+	return newslot;
+}
+
+static bool
+fsm_addr_on_same_page(FSMAddress addr1, FSMAddress addr2)
+{
+	if ((addr1.level != addr2.level)
+		|| (addr1.logpageno != addr2.logpageno))
+		return false;
+
+	return true;
+}
+
+/*
  * Search the tree for a heap page with at least min_cat of free space
  */
 static BlockNumber
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 0632fc0..7e04137 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -341,6 +341,41 @@ LockRelationForExtension(Relation relation, LOCKMODE lockmode)
 }
 
 /*
+ *		ConditionLockRelationForExtension
+ *
+ * As above, but only lock if we can get the lock without blocking.
+ * Returns TRUE iff the lock was acquired.
+ */
+bool
+ConditionLockRelationForExtension(Relation relation, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return (LockAcquire(&tag, lockmode, false, true) != LOCKACQUIRE_NOT_AVAIL);
+}
+
+/*
+ *		RelationExtensionLockWaiterCount
+ *
+ * Count the lock requester for the RelationExtension lock.
+ */
+int
+RelationExtensionLockWaiterCount(Relation relation)
+{
+	LOCKTAG		tag;
+
+	SET_LOCKTAG_RELATION_EXTEND(tag,
+								relation->rd_lockInfo.lockRelId.dbId,
+								relation->rd_lockInfo.lockRelId.relId);
+
+	return LockWaiterCount(&tag);
+}
+
+/*
  *		UnlockRelationForExtension
  */
 void
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index b30b7b1..353f705 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -4380,3 +4380,40 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	LockRelease(&tag, ShareLock, false);
 	return true;
 }
+
+/*
+ * LockWaiterCount
+ *
+ * Find the number of lock requester on this locktag
+ */
+int
+LockWaiterCount(const LOCKTAG *locktag)
+{
+	LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
+	LOCK	   *lock;
+	bool		found;
+	uint32		hashcode;
+	LWLock		*partitionLock;
+	int			waiters = 0;
+
+	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
+		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
+
+	hashcode = LockTagHashCode(locktag);
+	partitionLock = LockHashPartitionLock(hashcode);
+	LWLockAcquire(partitionLock, LW_EXCLUSIVE);
+
+	lock = (LOCK *) hash_search_with_hash_value(LockMethodLockHash,
+												(const void *) locktag,
+												hashcode,
+												HASH_FIND,
+												&found);
+	if (found)
+	{
+		Assert(lock != NULL);
+		waiters = lock->nRequested;
+	}
+	LWLockRelease(partitionLock);
+
+	return waiters;
+}
diff --git a/src/include/storage/freespace.h b/src/include/storage/freespace.h
index 19dcb8d..281137a 100644
--- a/src/include/storage/freespace.h
+++ b/src/include/storage/freespace.h
@@ -32,5 +32,10 @@ extern void XLogRecordPageWithFreeSpace(RelFileNode rnode, BlockNumber heapBlk,
 
 extern void FreeSpaceMapTruncateRel(Relation rel, BlockNumber nblocks);
 extern void FreeSpaceMapVacuum(Relation rel);
+extern BlockNumber GetPageWithFreeSpaceExtended(Relation rel,
+								BlockNumber oldPage,
+								BlockNumber newPage,
+								Size spaceNeeded);
+
 
 #endif   /* FREESPACE_H_ */
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index 975b6f8..4460756 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -53,6 +53,9 @@ extern void UnlockRelationIdForSession(LockRelId *relid, LOCKMODE lockmode);
 /* Lock a relation for extension */
 extern void LockRelationForExtension(Relation relation, LOCKMODE lockmode);
 extern void UnlockRelationForExtension(Relation relation, LOCKMODE lockmode);
+extern bool ConditionLockRelationForExtension(Relation relation,
+								  LOCKMODE lockmode);
+extern int	RelationExtensionLockWaiterCount(Relation relation);
 
 /* Lock a page (currently only used within indexes) */
 extern void LockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode);
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index b26427d..9c08679 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -574,6 +574,8 @@ extern void RememberSimpleDeadLock(PGPROC *proc1,
 					   PGPROC *proc2);
 extern void InitDeadLockChecking(void);
 
+extern int LockWaiterCount(const LOCKTAG *locktag);
+
 #ifdef LOCK_DEBUG
 extern void DumpLocks(PGPROC *proc);
 extern void DumpAllLocks(void);