Perform inmem_smgr cleaup after processing each record (#154)

* Perform inmem_smgr cleaup after processing each record

* Prevent eviction of wal redo target page

* Prevent eviction of wal redo target page frmo temp buffers

Author: knizhnik

contrib/zenith/inmem_smgr.c

@@ -28,7 +28,7 @@
 #include "storage/relfilenode.h"
 #include "storage/smgr.h"
 
-#define MAX_PAGES 128
+#define MAX_PAGES 32
 
 static BufferTag page_tag[MAX_PAGES];
 static char page_body[MAX_PAGES][BLCKSZ];
@@ -195,10 +195,6 @@ inmem_nblocks(SMgrRelation reln, ForkNumber forknum)
 {
 	int			nblocks = 0;
 
-	/*
-	 * Find the hightest-numbered page, and report that as the relation size.
-	 * XXX: Why does this get called during WAL replay at all?
-	 */
 	for (int i = 0; i < used_pages; i++)
 	{
 		if (RelFileNodeEquals(reln->smgr_rnode.node, page_tag[i].rnode)

src/backend/storage/buffer/localbuf.c

@@ -25,6 +25,8 @@
 #include "utils/memutils.h"
 #include "utils/resowner_private.h"
 
+/* ZENITH: prevent eviction of the buffer of target page */
+extern Buffer wal_redo_buffer;
 
 /*#define LBDEBUG*/
 
@@ -183,6 +185,12 @@ LocalBufferAlloc(SMgrRelation smgr, ForkNumber forkNum, BlockNumber blockNum,
 
 		if (LocalRefCount[b] == 0)
 		{
+			if (-b - 1 == wal_redo_buffer)
+			{
+				/* ZENITH: Prevent eviction of the buffer with target wal redo page */
+				continue;
+			}
+
 			buf_state = pg_atomic_read_u32(&bufHdr->state);
 
 			if (BUF_STATE_GET_USAGECOUNT(buf_state) > 0)

src/backend/tcop/zenith_wal_redo.c

@@ -100,6 +100,7 @@ static ssize_t buffered_read(void *buf, size_t count);
 
 static BufferTag target_redo_tag;
 
+Buffer		wal_redo_buffer;
 bool		am_wal_redo_postgres;
 
 static XLogReaderState *reader_state;
@@ -566,6 +567,7 @@ PushPage(StringInfo input_message)
 	content = pq_getmsgbytes(input_message, BLCKSZ);
 
 	buf = ReadBufferWithoutRelcache(rnode, forknum, blknum, RBM_ZERO_AND_LOCK, NULL);
+	wal_redo_buffer = buf;
 	page = BufferGetPage(buf);
 	memcpy(page, content, BLCKSZ);
 	MarkBufferDirty(buf); /* pro forma */
@@ -594,6 +596,8 @@ ApplyRecord(StringInfo input_message)
 	 */
 	lsn = pq_getmsgint64(input_message);
 
+	smgrinit();					/* reset inmem smgr state */
+
 	/* note: the input must be aligned here */
 	record = (XLogRecord *) pq_getmsgbytes(input_message, sizeof(XLogRecord));
 
@@ -734,7 +738,6 @@ GetPage(StringInfo input_message)
 
 	ReleaseBuffer(buf);
 	DropRelFileNodeAllLocalBuffers(rnode);
-	smgrinit();					/* reset inmem smgr state */
 
 	elog(TRACE, "Page sent back for block %u", blknum);
 }