Support snapshots at replica

Author: knizhnik

src/backend/access/rmgrdesc/standbydesc.c

@@ -84,6 +84,15 @@ standby_identify(uint8 info)
 		case XLOG_INVALIDATIONS:
 			id = "INVALIDATIONS";
 			break;
+		case XLOG_MAKE_SNAPSHOT:
+			id = "MAKE_SNAPSHOT";
+			break;
+		case XLOG_REMOVE_SNAPSHOT:
+			id = "REMOVE_SNAPSHOT";
+			break;
+		case XLOG_RECOVER_TO_SNAPSHOT:
+			id = "RECOVER_TO_SNAPSHOT";
+			break;
 	}
 
 	return id;

src/backend/storage/file/fd.c

@@ -91,6 +91,7 @@
 #include "storage/fd.h"
 #include "storage/ipc.h"
 #include "storage/snapfs.h"
+#include "storage/standbydefs.h"
 #include "utils/guc.h"
 #include "utils/inval.h"
 #include "utils/resowner_private.h"
@@ -4021,6 +4022,8 @@ sfs_remove_snapshot(SnapshotId snap_id)
 
 	ControlFile->oldest_snapshot = snap_id + 1;
 	UpdateControlFile();
+
+	sfs_xlog_insert(XLOG_REMOVE_SNAPSHOT, snap_id);
 }
 
 void
@@ -4032,9 +4035,7 @@ sfs_recover_to_snapshot(SnapshotId snap_id)
 		elog(ERROR, "Can not perform operation inside snapshot");
 
 	sfs_lock_database();
-
-	RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT
-					  | CHECKPOINT_FLUSH_ALL);
+	sfs_checkpoint();
 
 	sfs_current_snapshot = snap_id;
 	walk_data_dir(sfs_recover_snapshot_file, ERROR);
@@ -4046,8 +4047,12 @@ sfs_recover_to_snapshot(SnapshotId snap_id)
 	walk_data_dir(sfs_remove_applied_snapshot_file, LOG);
 
 	DropSharedBuffers();
-	InvalidateSystemCaches();
-	CacheInvalidateRelcacheAll();
-
+	if (!InRecovery)
+	{
+		InvalidateSystemCaches();
+		CacheInvalidateRelcacheAll();
+	}
 	sfs_unlock_database();
+
+	sfs_xlog_insert(XLOG_RECOVER_TO_SNAPSHOT, snap_id);
 }

src/backend/storage/file/snapfs.c

@@ -33,6 +33,7 @@
 #include "storage/lwlock.h"
 #include "storage/procarray.h"
 #include "storage/snapfs.h"
+#include "storage/standbydefs.h"
 #include "storage/bufmgr.h"
 #include "utils/inval.h"
 #include "utils/fmgrprotos.h"
@@ -163,8 +164,7 @@ sfs_switch_to_snapshot(SnapshotId snap_id)
 	sfs_lock_database();
 
 	if (!SFS_IN_SNAPSHOT())
-		RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT
-						  | CHECKPOINT_FLUSH_ALL);
+		sfs_checkpoint();
 
 	ControlFile->active_snapshot = snap_id;
 	UpdateControlFile();
@@ -197,14 +197,15 @@ sfs_make_snapshot(void)
 		elog(ERROR, "Can not perform operation inside snapshot");
 
 	sfs_lock_database();
+	sfs_checkpoint();
 
-	RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT
-					  | CHECKPOINT_FLUSH_ALL);
 	snap_id = ++ControlFile->recent_snapshot;
 	UpdateControlFile();
 
 	sfs_unlock_database();
 
+	sfs_xlog_insert(XLOG_MAKE_SNAPSHOT, snap_id);
+
 	return snap_id;
 }
 
@@ -214,34 +215,66 @@ sfs_make_snapshot(void)
 void
 sfs_lock_database(void)
 {
-	bool standalone = false;
-	TransactionId myXid = GetCurrentTransactionIdIfAny();
-	/* Prevent assignment Xids to transaction and
-	 * so delay start of any new update transactions
-	 */
-	LWLockAcquire(XidGenLock, LW_SHARED);
-
-	/* Wait completion of all active tranasction except own */
-	do
+	if (!InRecovery)
 	{
-		RunningTransactions running = GetRunningTransactionData();
-		standalone = (TransactionIdIsValid(myXid) && running->xcnt == 1) || (!TransactionIdIsValid(myXid) && running->xcnt == 0);
+		bool standalone = false;
+		TransactionId myXid = GetCurrentTransactionIdIfAny();
 
-		/* Release locks set by GetRunningTransactionData */
-		LWLockRelease(ProcArrayLock);
-		LWLockRelease(XidGenLock);
+		/* Prevent assignment Xids to transaction and
+		 * so delay start of any new update transactions
+		 */
+		LWLockAcquire(XidGenLock, LW_SHARED);
+
+		/* Wait completion of all active tranasction except own */
+		do
+		{
+			RunningTransactions running = GetRunningTransactionData();
+			standalone = (TransactionIdIsValid(myXid) && running->xcnt == 1) || (!TransactionIdIsValid(myXid) && running->xcnt == 0);
+
+			/* Release locks set by GetRunningTransactionData */
+			LWLockRelease(ProcArrayLock);
+			LWLockRelease(XidGenLock);
 
-		/* Wait for one second */
-		if (!standalone)
-			pg_usleep(USECS_PER_SEC);
+			/* Wait for one second */
+			if (!standalone)
+				pg_usleep(USECS_PER_SEC);
 
-	} while (!standalone);
+		} while (!standalone);
+	}
 }
 
 void
 sfs_unlock_database(void)
 {
-	LWLockRelease(XidGenLock);
+	if (!InRecovery)
+		LWLockRelease(XidGenLock);
+}
+
+void
+sfs_xlog_insert(int op, SnapshotId snapid)
+{
+	if (!InRecovery)
+	{
+		xl_snapshot xlrec;
+		xlrec.snapid = snapid;
+		XLogBeginInsert();
+		XLogSetRecordFlags(XLOG_MARK_UNIMPORTANT);
+		XLogRegisterData((char *) (&xlrec), sizeof(xlrec));
+		XLogInsert(RM_STANDBY_ID, op);
+	}
+}
+
+
+void
+sfs_checkpoint(void)
+{
+	int flags = CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT | CHECKPOINT_FLUSH_ALL;
+/*
+	if (InRecovery)
+		CreateRestartPoint(flags);
+	else
+*/
+	RequestCheckpoint(flags);
 }
 
 /*

src/backend/storage/ipc/standby.c

@@ -28,6 +28,7 @@
 #include "storage/proc.h"
 #include "storage/procarray.h"
 #include "storage/sinvaladt.h"
+#include "storage/snapfs.h"
 #include "storage/standby.h"
 #include "utils/hsearch.h"
 #include "utils/memutils.h"
@@ -827,6 +828,24 @@ standby_redo(XLogReaderState *record)
 											 xlrec->dbId,
 											 xlrec->tsId);
 	}
+	else if (info == XLOG_MAKE_SNAPSHOT)
+	{
+		xl_snapshot *xlrec = (xl_snapshot *) XLogRecGetData(record);
+		SnapshotId snapid = sfs_make_snapshot();
+		if (snapid != xlrec->snapid)
+			elog(WARNING, "Snapshot identifiers at replica and master do not match: %d vs. %d",
+				 snapid, xlrec->snapid);
+	}
+	else if (info == XLOG_RECOVER_TO_SNAPSHOT)
+	{
+		xl_snapshot *xlrec = (xl_snapshot *) XLogRecGetData(record);
+		sfs_recover_to_snapshot(xlrec->snapid);
+	}
+	else if (info == XLOG_REMOVE_SNAPSHOT)
+	{
+		xl_snapshot *xlrec = (xl_snapshot *) XLogRecGetData(record);
+		sfs_remove_snapshot(xlrec->snapid);
+	}
 	else
 		elog(PANIC, "standby_redo: unknown op code %u", info);
 }

src/include/catalog/pg_control.h

@@ -51,7 +51,7 @@ typedef struct CheckPoint
 	TransactionId oldestCommitTsXid;	/* oldest Xid with valid commit
 										 * timestamp */
 	TransactionId newestCommitTsXid;	/* newest Xid with valid commit
-										 * timestamp */
+													 * timestamp */
 
 	/*
 	 * Oldest XID still running. This is only needed to initialize hot standby

src/include/storage/snapfs.h

@@ -90,6 +90,9 @@ extern void sfs_unlock_database(void);
 
 extern void sfs_check_snapshot(SnapshotId sid);
 
+extern void sfs_xlog_insert(int op, SnapshotId snapid);
+extern void sfs_checkpoint(void);
+
 extern struct ControlFileData *ControlFile;
 
 #endif

src/include/storage/standbydefs.h

@@ -34,6 +34,9 @@ extern void standby_desc_invalidations(StringInfo buf,
 #define XLOG_STANDBY_LOCK			0x00
 #define XLOG_RUNNING_XACTS			0x10
 #define XLOG_INVALIDATIONS			0x20
+#define XLOG_MAKE_SNAPSHOT			0x30
+#define XLOG_REMOVE_SNAPSHOT	    0x40
+#define XLOG_RECOVER_TO_SNAPSHOT	0x50
 
 typedef struct xl_standby_locks
 {
@@ -69,6 +72,14 @@ typedef struct xl_invalidations
 	SharedInvalidationMessage msgs[FLEXIBLE_ARRAY_MEMBER];
 } xl_invalidations;
 
+/*
+ * Snapshot operation at standby
+ */
+typedef struct xl_snapshot
+{
+	int         snapid;
+}  xl_snapshot;
+
 #define MinSizeOfInvalidations offsetof(xl_invalidations, msgs)
 
 #endif							/* STANDBYDEFS_H */