Consistency improvements for slot and decoding code.

anarazel · anarazel · commit e04a9ccd2ccd · 2014-06-12T13:33:27.000+02:00
Change the order of checks in similar functions to be the same; remove
a parameter that's not needed anymore; rename a memory context and
expand a couple of comments.

Per review comments from Amit Kapila
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
@@ -6270,7 +6270,7 @@ StartupXLOG(void)
 	 * Initialize replication slots, before there's a chance to remove
 	 * required resources.
 	 */
-	StartupReplicationSlots(checkPoint.redo);
+	StartupReplicationSlots();
 
 	/*
 	 * Startup logical state, needs to be setup now so we have proper data
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
@@ -125,7 +125,7 @@ StartupDecodingContext(List *output_plugin_options,
 	slot = MyReplicationSlot;
 
 	context = AllocSetContextCreate(CurrentMemoryContext,
-									"Changeset Extraction Context",
+									"Logical Decoding Context",
 									ALLOCSET_DEFAULT_MINSIZE,
 									ALLOCSET_DEFAULT_INITSIZE,
 									ALLOCSET_DEFAULT_MAXSIZE);
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
@@ -829,7 +829,7 @@ CheckPointReplicationSlots(void)
  * needs to be run before we start crash recovery.
  */
 void
-StartupReplicationSlots(XLogRecPtr checkPointRedo)
+StartupReplicationSlots(void)
 {
 	DIR		   *replication_dir;
 	struct dirent *replication_de;
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
@@ -46,13 +46,15 @@ pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
 	HeapTuple	tuple;
 	Datum		result;
 
-	check_permissions();
-
-	CheckSlotRequirements();
+	Assert(!MyReplicationSlot);
 
 	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
 		elog(ERROR, "return type must be a row type");
 
+	check_permissions();
+
+	CheckSlotRequirements();
+
 	/* acquire replication slot, this will check for conflicting names */
 	ReplicationSlotCreate(NameStr(*name), false, RS_PERSISTENT);
 
@@ -87,17 +89,20 @@ pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
 	Datum		values[2];
 	bool		nulls[2];
 
+	Assert(!MyReplicationSlot);
+
 	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
 		elog(ERROR, "return type must be a row type");
 
 	check_permissions();
 
 	CheckLogicalDecodingRequirements();
 
-	Assert(!MyReplicationSlot);
-
 	/*
-	 * Acquire a logical decoding slot, this will check for conflicting names.
+	 * Acquire a logical decoding slot, this will check for conflicting
+	 * names. Initially create it as ephemeral - that allows us to nicely
+	 * handle errors during initialization because it'll get dropped if this
+	 * transaction fails. We'll make it persistent at the end.
 	 */
 	ReplicationSlotCreate(NameStr(*name), true, RS_EPHEMERAL);
 
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
@@ -781,6 +781,11 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 	else
 	{
 		CheckLogicalDecodingRequirements();
+		/*
+		 * Initially create the slot as ephemeral - that allows us to nicely
+		 * handle errors during initialization because it'll get dropped if
+		 * this transaction fails. We'll make it persistent at the end.
+		 */
 		ReplicationSlotCreate(cmd->slotname, true, RS_EPHEMERAL);
 	}
 
@@ -1682,8 +1687,8 @@ ProcessStandbyHSFeedbackMessage(void)
 	 * If we're using a replication slot we reserve the xmin via that,
 	 * otherwise via the walsender's PGXACT entry.
 	 *
-	 * XXX: It might make sense to introduce ephemeral slots and always use
-	 * the slot mechanism.
+	 * XXX: It might make sense to generalize the ephemeral slot concept and
+	 * always use the slot mechanism to handle the feedback xmin.
 	 */
 	if (MyReplicationSlot != NULL)		/* XXX: persistency configurable? */
 		PhysicalReplicationSlotNewXmin(feedbackXmin);
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
@@ -164,7 +164,7 @@ extern void ReplicationSlotsComputeRequiredLSN(void);
 extern XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void);
 extern bool ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive);
 
-extern void StartupReplicationSlots(XLogRecPtr checkPointRedo);
+extern void StartupReplicationSlots(void);
 extern void CheckPointReplicationSlots(void);
 
 extern void CheckSlotRequirements(void);

Original file line number	Diff line number	Diff line change
`@@ -829,7 +829,7 @@ CheckPointReplicationSlots(void)`
`829`	`829`	`* needs to be run before we start crash recovery.`
`830`	`830`	`*/`
`831`	`831`	`void`
`832`		`-StartupReplicationSlots(XLogRecPtr checkPointRedo)`
	`832`	`+StartupReplicationSlots(void)`
`833`	`833`	`{`
`834`	`834`	`DIR *replication_dir;`
`835`	`835`	`struct dirent *replication_de;`