An attempt to reduce WALWriteLock contention

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index c1b9a97..fc79415 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -73,6 +73,7 @@
 #include "utils/snapmgr.h"
 #include "utils/timestamp.h"
 #include "pg_trace.h"
+#include "access/rmgr.h"
 
 extern uint32 bootstrap_data_checksum_version;
 
@@ -808,7 +809,8 @@ static XLogRecPtr XLogGetReplicationSlotMinimumLSN(void);
 
 static void AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic);
 static bool XLogCheckpointNeeded(XLogSegNo new_segno);
-static void XLogWrite(XLogwrtRqst WriteRqst, bool flexible);
+static void XLogWrite(XLogwrtRqst WriteRqst, bool flexible, PGPROC_LIST *wake_pendingWriteWALElem);
+static void XLogFsync(XLogwrtRqst WriteRqst, bool flexible);
 static bool InstallXLogFileSegment(XLogSegNo *segno, char *tmppath,
 					   bool find_free, XLogSegNo max_segno,
 					   bool use_lock);
@@ -1917,7 +1919,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic)
 
 				LWLockAcquire(WALWriteLock, LW_EXCLUSIVE);
 
-				LogwrtResult = XLogCtl->LogwrtResult;
+				LogwrtResult.Write = XLogCtl->LogwrtResult.Write;
 				if (LogwrtResult.Write >= OldPageRqstPtr)
 				{
 					/* OK, someone wrote it already */
@@ -1929,8 +1931,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic)
 					TRACE_POSTGRESQL_WAL_BUFFER_WRITE_DIRTY_START();
 					WriteRqst.Write = OldPageRqstPtr;
 					WriteRqst.Flush = 0;
-					XLogWrite(WriteRqst, false);
-					LWLockRelease(WALWriteLock);
+					XLogWrite(WriteRqst, false, NULL);
 					TRACE_POSTGRESQL_WAL_BUFFER_WRITE_DIRTY_DONE();
 				}
 				/* Re-acquire WALBufMappingLock and retry */
@@ -2152,7 +2153,7 @@ XLogCheckpointNeeded(XLogSegNo new_segno)
  * write.
  */
 static void
-XLogWrite(XLogwrtRqst WriteRqst, bool flexible)
+XLogWrite(XLogwrtRqst WriteRqst, bool flexible, PGPROC_LIST *wake_pendingWriteWALElem)
 {
 	bool		ispartialpage;
 	bool		last_iteration;
@@ -2162,6 +2163,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible)
 	int			npages;
 	int			startidx;
 	uint32		startoffset;
+	long long	start;
+	bool		moveToNextBuf;
+	PGPROC      *proc_to_clear;
+
 
 	/* We should always be inside a critical section here */
 	Assert(CritSectionCount > 0);
@@ -2169,7 +2174,7 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible)
 	/*
 	 * Update local LogwrtResult (caller probably did this already, but...)
 	 */
-	LogwrtResult = XLogCtl->LogwrtResult;
+	LogwrtResult.Write = XLogCtl->LogwrtResult.Write;
 
 	/*
 	 * Since successive pages in the xlog cache are consecutively allocated,
@@ -2183,6 +2188,7 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible)
 	npages = 0;
 	startidx = 0;
 	startoffset = 0;
+	moveToNextBuf = true;
 
 	/*
 	 * Within the loop, curridx is the cache block index of the page to
@@ -2193,18 +2199,35 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible)
 
 	while (LogwrtResult.Write < WriteRqst.Write)
 	{
+
+		XLogRecPtr  EndPtr;
+
+		if (!moveToNextBuf)
+		{
+			/*
+			 * Within the loop, curridx is the cache block index of the page to
+			 * consider writing.  Begin at the buffer containing the next unwritten
+			 * page, or last partially written page.
+			 */
+			curridx = XLogRecPtrToBufIdx(LogwrtResult.Write);
+			moveToNextBuf = true;
+		}
+
 		/*
 		 * Make sure we're not ahead of the insert process.  This could happen
 		 * if we're passed a bogus WriteRqst.Write that is past the end of the
 		 * last page that's been initialized by AdvanceXLInsertBuffer.
 		 */
-		XLogRecPtr	EndPtr = XLogCtl->xlblocks[curridx];
+		EndPtr = XLogCtl->xlblocks[curridx];
 
 		if (LogwrtResult.Write >= EndPtr)
-			elog(PANIC, "xlog write request %X/%X is past end of log %X/%X",
+		{
+			elog(LOG, "xlog write request %X/%X is past end of log %X/%X",
 				 (uint32) (LogwrtResult.Write >> 32),
 				 (uint32) LogwrtResult.Write,
 				 (uint32) (EndPtr >> 32), (uint32) EndPtr);
+			break;
+		}
 
 		/* Advance LogwrtResult.Write to end of current buffer page */
 		LogwrtResult.Write = EndPtr;
@@ -2318,13 +2341,47 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible)
 			 */
 			if (finishing_seg)
 			{
+				/*
+				 * Update shared memory status to indicate write
+				 * progress as we are releasing WALWriteLock.
+				 */
+				SpinLockAcquire(&XLogCtl->info_lck);
+				XLogCtl->LogwrtResult.Write = LogwrtResult.Write;
+				if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write)
+					XLogCtl->LogwrtRqst.Write = LogwrtResult.Write;
+				SpinLockRelease(&XLogCtl->info_lck);
+
+				LWLockRelease(WALWriteLock);
+
+				while (wake_pendingWriteWALElem)
+				{
+					proc_to_clear = (PGPROC *) (((char *) wake_pendingWriteWALElem) -
+												offsetof(PGPROC, pendingWriteWALLinks));
+
+					wake_pendingWriteWALElem = wake_pendingWriteWALElem->next;
+
+					/* Mark that Xid has cleared for this proc */
+					proc_to_clear->writeWAL = false;
+
+					pg_write_barrier();
+
+					if (proc_to_clear != MyProc)
+						PGSemaphoreUnlock(&proc_to_clear->sem);
+				}
+
+
+				LWLockAcquire(WALFlushLock, LW_EXCLUSIVE);
+
+				/*
+				 * we want to ensure that current segment is completely
+				 * flushed.
+				 */
 				issue_xlog_fsync(openLogFile, openLogSegNo);
+				LogwrtResult.Flush = LogwrtResult.Write;		/* end of page */
 
 				/* signal that we need to wakeup walsenders later */
 				WalSndWakeupRequest();
 
-				LogwrtResult.Flush = LogwrtResult.Write;		/* end of page */
-
 				if (XLogArchivingActive())
 					XLogArchiveNotifySeg(openLogSegNo);
 
@@ -2343,6 +2400,30 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible)
 					if (XLogCheckpointNeeded(openLogSegNo))
 						RequestCheckpoint(CHECKPOINT_CAUSE_XLOG);
 				}
+				/*
+				 * Update shared memory status to indicate flush
+				 * progress as we are releasing WALFlushLock.
+				 */
+				SpinLockAcquire(&XLogCtl->info_lck);
+				if (XLogCtl->LogwrtResult.Flush < LogwrtResult.Flush)
+					XLogCtl->LogwrtResult.Flush = LogwrtResult.Flush;
+				if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush)
+					XLogCtl->LogwrtRqst.Flush = LogwrtResult.Flush;
+				LogwrtResult.Flush = XLogCtl->LogwrtResult.Flush;
+				SpinLockRelease(&XLogCtl->info_lck);
+
+				LWLockRelease(WALFlushLock);
+
+				/*
+				 * Reacquire the WALWriteLock and get the Write progress.
+				 * We don't need spinlocks since, we update write progress
+				 * only while holding WALWriteLock.
+				 */
+				LWLockAcquire(WALWriteLock, LW_EXCLUSIVE);
+				LogwrtResult.Write = XLogCtl->LogwrtResult.Write;
+
+				/* Next cache block may have been writtent already. */
+				moveToNextBuf = false;
 			}
 		}
 
@@ -2362,6 +2443,40 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible)
 	Assert(npages == 0);
 
 	/*
+	 * Update shared memory status to indicate write
+	 * progress as we are releasing WALWriteLock. Also,
+	 * update local copy og LogwrtResult since we got a chance.
+	 */
+	SpinLockAcquire(&XLogCtl->info_lck);
+	if (XLogCtl->LogwrtResult.Write < LogwrtResult.Write)
+		XLogCtl->LogwrtResult.Write = LogwrtResult.Write;
+	if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write)
+		XLogCtl->LogwrtRqst.Write = LogwrtResult.Write;
+	LogwrtResult = XLogCtl->LogwrtResult;
+	SpinLockRelease(&XLogCtl->info_lck);
+	LWLockRelease(WALWriteLock);
+
+	/*
+	 * Now wake up all the procs waiting for their writes to be
+	 * completed.
+	 */
+	while (wake_pendingWriteWALElem)
+	{
+		proc_to_clear = (PGPROC *) (((char *) wake_pendingWriteWALElem) -
+									offsetof(PGPROC, pendingWriteWALLinks));
+
+		wake_pendingWriteWALElem = wake_pendingWriteWALElem->next;
+
+		/* Mark that Xid has cleared for this proc */
+		proc_to_clear->writeWAL = false;
+
+		pg_write_barrier();
+
+		if (proc_to_clear != MyProc)
+			PGSemaphoreUnlock(&proc_to_clear->sem);
+	}
+
+	/*
 	 * If asked to flush, do so
 	 */
 	if (LogwrtResult.Flush < WriteRqst.Flush &&
@@ -2376,40 +2491,71 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible)
 		if (sync_method != SYNC_METHOD_OPEN &&
 			sync_method != SYNC_METHOD_OPEN_DSYNC)
 		{
-			if (openLogFile >= 0 &&
-				!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo))
-				XLogFileClose();
-			if (openLogFile < 0)
+			for(;;)
 			{
-				XLByteToPrevSeg(LogwrtResult.Write, openLogSegNo);
-				openLogFile = XLogFileOpen(openLogSegNo);
-				openLogOff = 0;
-			}
+				SpinLockAcquire(&XLogCtl->info_lck);
+				LogwrtResult = XLogCtl->LogwrtResult;
+				SpinLockRelease(&XLogCtl->info_lck);
 
-			issue_xlog_fsync(openLogFile, openLogSegNo);
-		}
+				/* done already? */
+				if (WriteRqst.Flush <= LogwrtResult.Flush ||
+					LogwrtResult.Write <= LogwrtResult.Flush)
+					break;
 
-		/* signal that we need to wakeup walsenders later */
-		WalSndWakeupRequest();
+				if (!LWLockAcquireOrWait(WALFlushLock, LW_EXCLUSIVE))
+				{
+					/*
+					 * The lock is now free, but we didn't acquire it yet. Before we
+					 * do, loop back to check if someone else flushed the record for
+					 * us already.
+					 */
+					continue;
+				}
+				if (enableFsync &&
+					MinimumActiveBackends(CommitSiblings))
+				{
+					//pg_usleep(1);
+				}
+				/* Got the lock; recheck whether request is satisfied */
+				LogwrtResult.Flush = XLogCtl->LogwrtResult.Flush;
 
-		LogwrtResult.Flush = LogwrtResult.Write;
-	}
+				if (WriteRqst.Flush <= LogwrtResult.Flush ||
+					LogwrtResult.Write <= LogwrtResult.Flush)
+				{
+					LWLockRelease(WALFlushLock);
+					break;
+				}
+				if (openLogFile >= 0 &&
+					!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo))
+				{
+					XLogFileClose();
+				}
+				if (openLogFile < 0)
+				{
+					XLByteToPrevSeg(LogwrtResult.Write, openLogSegNo);
+					openLogFile = XLogFileOpen(openLogSegNo);
+					openLogOff = 0;
+				}
 
-	/*
-	 * Update shared-memory status
-	 *
-	 * We make sure that the shared 'request' values do not fall behind the
-	 * 'result' values.  This is not absolutely essential, but it saves some
-	 * code in a couple of places.
-	 */
-	{
-		SpinLockAcquire(&XLogCtl->info_lck);
-		XLogCtl->LogwrtResult = LogwrtResult;
-		if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write)
-			XLogCtl->LogwrtRqst.Write = LogwrtResult.Write;
-		if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush)
-			XLogCtl->LogwrtRqst.Flush = LogwrtResult.Flush;
-		SpinLockRelease(&XLogCtl->info_lck);
+				issue_xlog_fsync(openLogFile, openLogSegNo);
+
+				LogwrtResult.Flush = LogwrtResult.Write;
+				/*
+				 * Update shared memory status to indicate flush
+				 * progress as we are releasing WALFlushLock.
+				 */
+				SpinLockAcquire(&XLogCtl->info_lck);
+				XLogCtl->LogwrtResult.Flush = LogwrtResult.Flush;
+				if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush)
+					XLogCtl->LogwrtRqst.Flush = LogwrtResult.Flush;
+				SpinLockRelease(&XLogCtl->info_lck);
+
+				/* signal that we need to wakeup walsenders later */
+				WalSndWakeupRequest();
+				LWLockRelease(WALFlushLock);
+				break;
+			}
+		}
 	}
 }
 
@@ -2573,6 +2719,15 @@ XLogFlush(XLogRecPtr record)
 {
 	XLogRecPtr	WriteRqstPtr;
 	XLogwrtRqst WriteRqst;
+	XLogRecPtr	insertpos;
+	/* use volatile pointer to prevent code rearrangement */
+	volatile PROC_HDR *procglobal = ProcGlobal;
+	PGPROC_LIST	*nonempty_pending_write_wal_list = NULL;
+	PGPROC_LIST	*pendingWriteWALElem, *wake_pendingWriteWALElem;
+	PGPROC		*proc_to_clear;
+	int			extraWaits = 0;
+	XLogRecPtr	largestWALWritePos = 0;
+	PGPROC		*proc = MyProc;
 
 	/*
 	 * During REDO, we are reading not writing WAL.  Therefore, instead of
@@ -2616,90 +2771,180 @@ XLogFlush(XLogRecPtr record)
 	 * Now wait until we get the write lock, or someone else does the flush
 	 * for us.
 	 */
-	for (;;)
+
+	/* read LogwrtResult and update local state */
+	SpinLockAcquire(&XLogCtl->info_lck);
+	if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write)
+		WriteRqstPtr = XLogCtl->LogwrtRqst.Write;
+	LogwrtResult = XLogCtl->LogwrtResult;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	/* done already? */
+	if (record <= LogwrtResult.Flush)
+		goto wal_is_written;
+
+	/*
+	 * Before actually performing the write, wait for all in-flight
+	 * insertions to the pages we're about to write to finish.
+	 */
+	insertpos = WaitXLogInsertionsToFinish(WriteRqstPtr);
+
+	/*
+	 * Except one proc, all other proc's will wait for their write location
+	 * to be written. Each proc will advertise it's write location and add
+	 * itself to the pendingWriteWALList and the first backend that has seen
+	 * the list as empty, will proceed to write the WAL for all the procs
+	 * in pendingWriteWALList (lets call it Write leader); if it is not the
+	 * first one, then just wait on semaphorelock till the wal is written for
+	 * the proc.  The Write leader will acquire WALWriteLock and traverse the
+	 * pendingWriteWALList to find the largest write location upto which it
+	 * needs to write the wal and use the same to write the WAL.  After write,
+	 * it will release the WALWriteLock and wakes all the procs for which it
+	 * has written the WAL.
+	 */
+	proc->writeWAL = true;
+
+	/*
+	 * Add the proc to pending pendingWriteWALList list and advertise
+	 * the insertpos, upto which the WAL needs to be written for this
+	 * proc.
+	 */
+	proc->writePos = insertpos;
+	while (true)
 	{
-		XLogRecPtr	insertpos;
+		proc->pendingWriteWALLinks.next = procglobal->pendingWriteWALList;
+		pg_read_barrier();
+		nonempty_pending_write_wal_list = proc->pendingWriteWALLinks.next;
+		if (pg_atomic_compare_exchange_u64((volatile pg_atomic_uint64*) &procglobal->pendingWriteWALList,
+											(uint64*)&proc->pendingWriteWALLinks.next,
+											(uint64)&proc->pendingWriteWALLinks))
+			break;
+	}
 
-		/* read LogwrtResult and update local state */
-		SpinLockAcquire(&XLogCtl->info_lck);
-		if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write)
-			WriteRqstPtr = XLogCtl->LogwrtRqst.Write;
-		LogwrtResult = XLogCtl->LogwrtResult;
-		SpinLockRelease(&XLogCtl->info_lck);
+	/*
+	 * only first process which has seen the pending write wal list as
+	 * empty will group write wal for proc's on pending list, all other
+	 * processes will wait for their wal to be written. Once, there wal
+	 * is written, they will try to acquire WALFlushLock to flush their
+	 * data. Here also, flush calls will be accumulated.
+	 */
+	if (nonempty_pending_write_wal_list)
+	{
+		for (;;)
+		{
+			PGSemaphoreLock(&proc->sem);
+			if (!proc->writeWAL)
+			{
+				/* logic similar to lwlock.c is used for any absorbed wakeups. */
+				while (extraWaits-- > 0)
+					PGSemaphoreUnlock(&proc->sem);
+				/*
+				 * Before exiting from here, read LogwrtResult and update
+				 * local state.  When the backend didn't write it's WAL, the
+				 * local state is not updated which we need to update to ensure
+				 * that WAL has been actually flushed, see the check in end of
+				 * this function.
+				 */
+				SpinLockAcquire(&XLogCtl->info_lck);
+				LogwrtResult = XLogCtl->LogwrtResult;
+				SpinLockRelease(&XLogCtl->info_lck);
+
+				if (record <= LogwrtResult.Flush)
+					goto wal_is_written;
+
+				while (true)
+				{
+					if (!LWLockAcquireOrWait(WALFlushLock, LW_EXCLUSIVE))
+					{
+						SpinLockAcquire(&XLogCtl->info_lck);
+						LogwrtResult = XLogCtl->LogwrtResult;
+						SpinLockRelease(&XLogCtl->info_lck);
+						if (record <= LogwrtResult.Flush)
+							goto wal_is_written;
+						continue;
+					}
+					if (record <= LogwrtResult.Flush)
+					{
+						LWLockRelease(WALFlushLock);
+						goto wal_is_written;
+					}
+					else
+					{
+						WriteRqst.Flush = record;
+						XLogFsync(WriteRqst, false);
+						goto wal_is_written;
+					}
+				}
+			}
+			extraWaits++;
+		}
+	}
 
-		/* done already? */
-		if (record <= LogwrtResult.Flush)
+	LWLockAcquire(WALWriteLock, LW_EXCLUSIVE);
+
+	while (true)
+	{
+		pendingWriteWALElem = procglobal->pendingWriteWALList;
+		if (pg_atomic_compare_exchange_u64((volatile pg_atomic_uint64*) &procglobal->pendingWriteWALList,
+											(uint64*)&pendingWriteWALElem,
+											(uint64)NULL))
 			break;
+	}
+
+	/* save the list of procs whose wal needs to be written to wake them up. */
+	wake_pendingWriteWALElem = pendingWriteWALElem;
+
+	while (pendingWriteWALElem)
+	{
+		proc_to_clear = (PGPROC *) (((char *) pendingWriteWALElem) -
+									offsetof(PGPROC, pendingWriteWALLinks));
 
 		/*
-		 * Before actually performing the write, wait for all in-flight
-		 * insertions to the pages we're about to write to finish.
+		 * remember the largest wal location upto which we need to
+		 * write.
 		 */
-		insertpos = WaitXLogInsertionsToFinish(WriteRqstPtr);
+		if (largestWALWritePos < proc_to_clear->writePos)
+			largestWALWritePos = proc_to_clear->writePos;
 
 		/*
-		 * Try to get the write lock. If we can't get it immediately, wait
-		 * until it's released, and recheck if we still need to do the flush
-		 * or if the backend that held the lock did it for us already. This
-		 * helps to maintain a good rate of group committing when the system
-		 * is bottlenecked by the speed of fsyncing.
+		 * move to next proc in list.
 		 */
-		if (!LWLockAcquireOrWait(WALWriteLock, LW_EXCLUSIVE))
-		{
-			/*
-			 * The lock is now free, but we didn't acquire it yet. Before we
-			 * do, loop back to check if someone else flushed the record for
-			 * us already.
-			 */
-			continue;
-		}
+		pendingWriteWALElem = pendingWriteWALElem->next;
+	}
 
-		/* Got the lock; recheck whether request is satisfied */
-		LogwrtResult = XLogCtl->LogwrtResult;
-		if (record <= LogwrtResult.Flush)
-		{
-			LWLockRelease(WALWriteLock);
-			break;
-		}
+	/* recheck whether request is satisfied, else write the wal. */
+	SpinLockAcquire(&XLogCtl->info_lck);
+	LogwrtResult = XLogCtl->LogwrtResult;
+	SpinLockRelease(&XLogCtl->info_lck);
 
-		/*
-		 * Sleep before flush! By adding a delay here, we may give further
-		 * backends the opportunity to join the backlog of group commit
-		 * followers; this can significantly improve transaction throughput,
-		 * at the risk of increasing transaction latency.
-		 *
-		 * We do not sleep if enableFsync is not turned on, nor if there are
-		 * fewer than CommitSiblings other backends with active transactions.
-		 */
-		if (CommitDelay > 0 && enableFsync &&
-			MinimumActiveBackends(CommitSiblings))
+	if (largestWALWritePos <= LogwrtResult.Flush)
+	{
+		LWLockRelease(WALWriteLock);
+		while (wake_pendingWriteWALElem)
 		{
-			pg_usleep(CommitDelay);
+			proc_to_clear = (PGPROC *) (((char *) wake_pendingWriteWALElem) -
+										offsetof(PGPROC, pendingWriteWALLinks));
 
-			/*
-			 * Re-check how far we can now flush the WAL. It's generally not
-			 * safe to call WaitXLogInsertionsToFinish while holding
-			 * WALWriteLock, because an in-progress insertion might need to
-			 * also grab WALWriteLock to make progress. But we know that all
-			 * the insertions up to insertpos have already finished, because
-			 * that's what the earlier WaitXLogInsertionsToFinish() returned.
-			 * We're only calling it again to allow insertpos to be moved
-			 * further forward, not to actually wait for anyone.
-			 */
-			insertpos = WaitXLogInsertionsToFinish(insertpos);
-		}
+			wake_pendingWriteWALElem = wake_pendingWriteWALElem->next;
 
-		/* try to write/flush later additions to XLOG as well */
-		WriteRqst.Write = insertpos;
-		WriteRqst.Flush = insertpos;
+			/* Mark that Xid has cleared for this proc */
+			proc_to_clear->writeWAL = false;
 
-		XLogWrite(WriteRqst, false);
+			pg_write_barrier();
 
-		LWLockRelease(WALWriteLock);
-		/* done */
-		break;
+			if (proc_to_clear != MyProc)
+				PGSemaphoreUnlock(&proc_to_clear->sem);
+		}
+	}
+	else
+	{
+		WriteRqst.Write = largestWALWritePos;
+		WriteRqst.Flush = largestWALWritePos;
+
+		XLogWrite(WriteRqst, false, wake_pendingWriteWALElem);
 	}
 
+wal_is_written:
 	END_CRIT_SECTION();
 
 	/* wake up walsenders now that we've released heavily contended locks */
@@ -2855,13 +3100,16 @@ XLogBackgroundFlush(void)
 	/* now wait for any in-progress insertions to finish and get write lock */
 	WaitXLogInsertionsToFinish(WriteRqst.Write);
 	LWLockAcquire(WALWriteLock, LW_EXCLUSIVE);
+	SpinLockAcquire(&XLogCtl->info_lck);
 	LogwrtResult = XLogCtl->LogwrtResult;
+	SpinLockRelease(&XLogCtl->info_lck);
 	if (WriteRqst.Write > LogwrtResult.Write ||
 		WriteRqst.Flush > LogwrtResult.Flush)
 	{
-		XLogWrite(WriteRqst, flexible);
+		XLogWrite(WriteRqst, flexible, NULL);
 	}
-	LWLockRelease(WALWriteLock);
+	else
+		LWLockRelease(WALWriteLock);
 
 	END_CRIT_SECTION();
 
@@ -11719,3 +11967,78 @@ XLogRequestWalReceiverReply(void)
 {
 	doRequestWalReceiverReply = true;
 }
+
+/*
+ * Similar to XLogWrite but will just fsync the wal pages
+ * in kernel buffers.
+ */
+static void
+XLogFsync(XLogwrtRqst WriteRqst, bool flexible)
+{
+	SpinLockAcquire(&XLogCtl->info_lck);
+	if (XLogCtl->LogwrtResult.Write < LogwrtResult.Write)
+		XLogCtl->LogwrtResult.Write = LogwrtResult.Write;
+	if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write)
+		XLogCtl->LogwrtRqst.Write = LogwrtResult.Write;
+	LogwrtResult = XLogCtl->LogwrtResult;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	if (LogwrtResult.Flush < WriteRqst.Flush)
+	{
+		/*
+		 * Could get here without iterating above loop, in which case we might
+		 * have no open file or the wrong one.  However, we do not need to
+		 * fsync more than one file.
+		 */
+		if (sync_method != SYNC_METHOD_OPEN &&
+			sync_method != SYNC_METHOD_OPEN_DSYNC)
+		{
+			for(;;)
+			{
+				SpinLockAcquire(&XLogCtl->info_lck);
+				LogwrtResult = XLogCtl->LogwrtResult;
+				SpinLockRelease(&XLogCtl->info_lck);
+
+				/* done already? */
+				if (WriteRqst.Flush <= LogwrtResult.Flush)
+				{
+					LWLockRelease(WALFlushLock);
+					break;
+				}
+
+				if (openLogFile >= 0 &&
+					!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo))
+				{
+					XLogFileClose();
+				}
+
+				if (openLogFile < 0)
+				{
+					XLByteToPrevSeg(LogwrtResult.Write, openLogSegNo);
+					openLogFile = XLogFileOpen(openLogSegNo);
+					openLogOff = 0;
+				}
+
+				issue_xlog_fsync(openLogFile, openLogSegNo);
+
+				LogwrtResult.Flush = WriteRqst.Flush;
+				/*
+				 * Update shared memory status to indicate flush
+				 * progress as we are releasing WALFlushLock.
+				 */
+				SpinLockAcquire(&XLogCtl->info_lck);
+				XLogCtl->LogwrtResult.Flush = LogwrtResult.Flush;
+				if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush)
+					XLogCtl->LogwrtRqst.Flush = LogwrtResult.Flush;
+				SpinLockRelease(&XLogCtl->info_lck);
+
+				/* signal that we need to wakeup walsenders later */
+				WalSndWakeupRequest();
+				LWLockRelease(WALFlushLock);
+				break;
+			}
+		}
+	}
+	else
+		LWLockRelease(WALFlushLock);
+}
diff --git a/src/backend/storage/lmgr/lwlocknames.txt b/src/backend/storage/lmgr/lwlocknames.txt
index f8996cd..82fec8a 100644
--- a/src/backend/storage/lmgr/lwlocknames.txt
+++ b/src/backend/storage/lmgr/lwlocknames.txt
@@ -47,3 +47,4 @@ CommitTsLock						39
 ReplicationOriginLock				40
 MultiXactTruncationLock				41
 OldSnapshotTimeMapLock				42
+WALFlushLock						43
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index 33e7023..0203b90 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -266,6 +266,8 @@ InitProcGlobal(void)
 		dlist_init(&procs[i].lockGroupMembers);
 	}
 
+	ProcGlobal->pendingWriteWALList = NULL;
+
 	/*
 	 * Save pointers to the blocks of PGPROC structures reserved for auxiliary
 	 * processes and prepared transactions.
@@ -405,6 +407,11 @@ InitProcess(void)
 	/* Initialize wait event information. */
 	MyProc->wait_event_info = 0;
 
+	/* Initialize fields for writting WAL */
+	MyProc->writeWAL = false;
+	MyProc->writePos = 0;
+	MyProc->pendingWriteWALLinks.next = NULL;
+
 	/*
 	 * Acquire ownership of the PGPROC's latch, so that we can use WaitLatch
 	 * on it.  That allows us to repoint the process latch, which so far
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index f576f05..d69cc33 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -66,6 +66,17 @@ struct XidCache
 #define INVALID_PGPROCNO		PG_INT32_MAX
 
 /*
+ * List to link PGPROC structures for the cases when they
+ * can't be linked via SHM_QUEUE like when we want to operate
+ * the list using atomic operations.  Currently this is used
+ * to link PGPROC's for clearing their XID information.
+ */
+typedef struct PGPROC_LIST
+{
+	struct PGPROC_LIST *next;
+} PGPROC_LIST;
+
+/*
  * Each backend has a PGPROC struct in shared memory.  There is also a list of
  * currently-unused PGPROC structs that will be reallocated to new backends.
  *
@@ -141,6 +152,9 @@ struct PGPROC
 	SHM_QUEUE	myProcLocks[NUM_LOCK_PARTITIONS];
 
 	struct XidCache subxids;	/* cache for subtransaction XIDs */
+	bool            writeWAL;
+	XLogRecPtr      writePos;
+	PGPROC_LIST     pendingWriteWALLinks;
 
 	/* Support for group XID clearing. */
 	/* true, if member of ProcArray group waiting for XID clear */
@@ -225,6 +239,8 @@ typedef struct PROC_HDR
 	PGPROC	   *autovacFreeProcs;
 	/* Head of list of bgworker free PGPROC structures */
 	PGPROC	   *bgworkerFreeProcs;
+	/* list of PGPROC structures that need to have their WAL written */
+	PGPROC_LIST *pendingWriteWALList;
 	/* First pgproc waiting for group XID clear */
 	pg_atomic_uint32 procArrayGroupFirst;
 	/* WALWriter process's latch */