intermediate wraparound

Author: Daniel Shelepanov

engine.c

@@ -353,6 +353,10 @@ ptrackCheckpoint(void)
 	struct stat stat_buf;
 	uint64		i = 0;
 	uint64		j = 0;
+	XLogRecPtr	new_init_lsn;
+	uint32		new_init_lsn32;
+	uint32		latest_lsn;
+	bool		lsn_was_advanced = false;
 
 	elog(DEBUG1, "ptrack checkpoint");
 
@@ -408,21 +412,25 @@ ptrackCheckpoint(void)
 	ptrack_write_chunk(ptrack_tmp_fd, &crc, (char *) ptrack_map,
 					   offsetof(PtrackMapHdr, init_lsn));
 
+	latest_lsn = pg_atomic_read_u32(&ptrack_map->latest_lsn);
 	init_lsn = pg_atomic_read_u32(&ptrack_map->init_lsn);
 
 	/* Set init_lsn during checkpoint if it is not set yet */
 	if (init_lsn == InvalidXLogRecPtr)
 	{
-		XLogRecPtr	new_init_lsn;
-		uint32		new_init_lsn32;
-
 		if (RecoveryInProgress())
 			new_init_lsn = GetXLogReplayRecPtr(NULL);
 		else
 			new_init_lsn = GetXLogInsertRecPtr();
 
 		new_init_lsn32 = (uint32)(new_init_lsn >> 16);
-
+		pg_atomic_write_u32(&ptrack_map->init_lsn, new_init_lsn32);
+		init_lsn = new_init_lsn32;
+	}
+	else if (lsn_diff(lsn_advance(init_lsn, PtrackLSNGap), latest_lsn) < 0)
+	{
+		new_init_lsn32 = lsn_advance(init_lsn, PtrackLSNGap);
+		lsn_was_advanced = true;
 		pg_atomic_write_u32(&ptrack_map->init_lsn, new_init_lsn32);
 		init_lsn = new_init_lsn32;
 	}
@@ -449,7 +457,11 @@ ptrackCheckpoint(void)
 		 * TODO: is it safe and can we do any better?
 		 */
 		lsn = pg_atomic_read_u32(&ptrack_map->entries[i]);
-		buf[j].value = lsn;
+
+		if (lsn_was_advanced && lsn_diff(lsn, init_lsn) < 0)
+			buf[j].value = InvalidXLogRecPtr;
+		else
+			buf[j].value = lsn;
 
 		i++;
 		j++;
@@ -467,7 +479,6 @@ ptrackCheckpoint(void)
 			ptrack_write_chunk(ptrack_tmp_fd, &crc, (char *) buf, writesz);
 			elog(DEBUG5, "ptrack checkpoint: i " UINT64_FORMAT ", j " UINT64_FORMAT ", writesz %zu PtrackContentNblocks " UINT64_FORMAT,
 				 i, j, writesz, (uint64) PtrackContentNblocks);
-
 			j = 0;
 		}
 	}
@@ -726,7 +737,12 @@ ptrack_mark_block(RelFileNodeBackend smgr_rnode,
 			   !pg_atomic_compare_exchange_u32(&ptrack_map->init_lsn, (uint32 *) &old_init_lsn.value, new_lsn32));
 	}
 
-	/* Atomically assign new LSN value to the first slot */
+	/* Assign latest_lsn first */
+	old_lsn.value = pg_atomic_read_u32(&ptrack_map->latest_lsn);
+	while (old_lsn.value < new_lsn32 &&
+		   !pg_atomic_compare_exchange_u32(&ptrack_map->latest_lsn, (uint32 *) &old_lsn.value, new_lsn32));
+
+	/* Then, atomically assign new LSN value to the first slot */
 	old_lsn.value = pg_atomic_read_u32(&ptrack_map->entries[slot1]);
 	elog(DEBUG3, "ptrack_mark_block: map[%zu]=%u <- %u", slot1, old_lsn.value, new_lsn32);
 	while (old_lsn.value < new_lsn32 &&

engine.h

@@ -65,6 +65,8 @@ typedef struct PtrackMapHdr
 	 */
 	uint32		version_num;
 
+	/* LSN of current writing position */
+	pg_atomic_uint32 latest_lsn;
 	/* LSN of the moment, when map was last enabled. */
 	pg_atomic_uint32 init_lsn;
 
@@ -94,6 +96,7 @@ typedef PtrackMapHdr * PtrackMap;
 #define BID_HASH_FUNC(bid) \
 		(DatumGetUInt64(hash_any_extended((unsigned char *)&bid, sizeof(bid), 0)))
 
+#define PtrackLSNGap	10e8
 /*
  * Per process pointer to shared ptrack_map
  */

ptrack.c

@@ -533,6 +533,8 @@ ptrack_get_pagemapset(PG_FUNCTION_ARGS)
 	datapagemap_t pagemap;
 	int64		pagecount = 0;
 	char		gather_path[MAXPGPATH];
+	uint32		init_lsn = InvalidXLogRecPtr;
+	bool		within_ptrack_map = true;
 
 	/* Exit immediately if there is no map */
 	if (ptrack_map == NULL)
@@ -639,6 +641,7 @@ ptrack_get_pagemapset(PG_FUNCTION_ARGS)
 				SRF_RETURN_DONE(funcctx);
 		}
 
+		init_lsn = pg_atomic_read_u32(&ptrack_map->init_lsn);
 		hash = BID_HASH_FUNC(ctx->bid);
 		slot1 = (size_t)(hash % PtrackContentNblocks);
 
@@ -649,8 +652,11 @@ ptrack_get_pagemapset(PG_FUNCTION_ARGS)
 				 (uint16) (update_lsn1 >> 16), (uint16) update_lsn1,
 				 ctx->bid.blocknum, ctx->relpath);
 
+		if (init_lsn != InvalidXLogRecPtr)
+			within_ptrack_map = lsn_diff(init_lsn, update_lsn1) <= 0;
+
 		/* Only probe the second slot if the first one is marked */
-		if (update_lsn1 >= ctx->lsn)
+		if (within_ptrack_map && lsn_diff(ctx->lsn, update_lsn1) <= 0)
 		{
 			slot2 = (size_t)(((hash << 32) | (hash >> 32)) % PtrackContentNblocks);
 			update_lsn2 = pg_atomic_read_u32(&ptrack_map->entries[slot2]);
@@ -660,8 +666,11 @@ ptrack_get_pagemapset(PG_FUNCTION_ARGS)
 					 (uint16) (update_lsn1 >> 16), (uint16) update_lsn2,
 					 ctx->bid.blocknum, ctx->relpath);
 
+			if (init_lsn != InvalidXLogRecPtr)
+				within_ptrack_map = lsn_diff(init_lsn, update_lsn2) <= 0;
+
 			/* Block has been changed since specified LSN.  Mark it in the bitmap */
-			if (update_lsn2 >= ctx->lsn)
+			if (within_ptrack_map && lsn_diff(ctx->lsn, update_lsn2) <= 0)
 			{
 				pagecount += 1;
 				datapagemap_add(&pagemap, ctx->bid.blocknum % ((BlockNumber) RELSEG_SIZE));

ptrack.h

@@ -47,6 +47,8 @@
 #define nodeOf(ndbck)		(ndbck).node
 #endif
 
+#define lsn_diff(lsn1, lsn2)	((int32)(lsn1-lsn2))
+#define lsn_advance(lsn, incr)	((uint32)(lsn+incr))
 /*
  * Structure identifying block on the disk.
  */