Make synchronous commit optional for apply

PJMODOS · PJMODOS · commit 53f713737cb5 · 2015-12-18T18:00:49.000+01:00
The default synchronous commit is now false.

We also have to keep track of commit lsn mapping between remote and local
so that we can report correct flush position to the provider.
diff --git a/pglogical.c b/pglogical.c
@@ -47,6 +47,8 @@ static const struct config_enum_entry PGLogicalConflictResolvers[] = {
 	{NULL, 0, false}
 };
 
+bool pglogical_synchronous_commit = false;
+
 void _PG_init(void);
 void pglogical_supervisor_main(Datum main_arg);
 
@@ -398,6 +400,14 @@ _PG_init(void)
 							 pglogical_conflict_resolver_check_hook,
 							 NULL, NULL);
 
+	DefineCustomBoolVariable("pglogical.synchronous_commit",
+							 "pglogical specific synchronous commit value",
+							 NULL,
+							 &pglogical_synchronous_commit,
+							 false, PGC_POSTMASTER,
+							 0,
+							 NULL, NULL, NULL);
+
 	if (IsBinaryUpgrade)
 		return;
 
diff --git a/pglogical.h b/pglogical.h
@@ -34,6 +34,8 @@
 
 #define REPLICATION_ORIGIN_ALL "all"
 
+extern bool pglogical_synchronous_commit;
+
 extern char *shorten_hash(const char *str, int maxlen);
 
 extern List *textarray_to_list(ArrayType *textarray);
diff --git a/pglogical_apply.c b/pglogical_apply.c
@@ -62,6 +62,7 @@
 #include "pglogical_worker.h"
 #include "pglogical.h"
 
+
 void pglogical_apply_main(Datum main_arg);
 
 static bool			in_remote_transaction = false;
@@ -77,6 +78,15 @@ PGLogicalSubscription	   *MySubscription = NULL;
 
 static PGconn	   *applyconn = NULL;
 
+typedef struct PGLFlushPosition
+{
+	dlist_node node;
+	XLogRecPtr local_end;
+	XLogRecPtr remote_end;
+} PGLFlushPosition;
+
+dlist_head lsn_mapping = DLIST_STATIC_INIT(lsn_mapping);
+
 static void handle_queued_message(HeapTuple msgtup, bool tx_just_started);
 static void handle_startup_param(const char *key, const char *value);
 static bool parse_bool_param(const char *key, const char *value);
@@ -144,7 +154,17 @@ handle_commit(StringInfo s)
 
 	if (IsTransactionState())
 	{
+		PGLFlushPosition *flushpos;
+
 		CommitTransactionCommand();
+		MemoryContextSwitchTo(TopMemoryContext);
+
+		/* Track commit lsn  */
+		flushpos = (PGLFlushPosition *) palloc(sizeof(PGLFlushPosition));
+		flushpos->local_end = XactLastCommitEnd;
+		flushpos->remote_end = end_lsn;
+
+		dlist_push_tail(&lsn_mapping, &flushpos->node);
 		MemoryContextSwitchTo(MessageContext);
 	}
 
@@ -1032,6 +1052,58 @@ replication_handler(StringInfo s)
 	}
 }
 
+/*
+ * Figure out which write/flush positions to report to the walsender process.
+ *
+ * We can't simply report back the last LSN the walsender sent us because the
+ * local transaction might not yet be flushed to disk locally. Instead we
+ * build a list that associates local with remote LSNs for every commit. When
+ * reporting back the flush position to the sender we iterate that list and
+ * check which entries on it are already locally flushed. Those we can report
+ * as having been flushed.
+ *
+ * Returns true if there's no outstanding transactions that need to be
+ * flushed.
+ */
+static bool
+get_flush_position(XLogRecPtr *write, XLogRecPtr *flush)
+{
+	dlist_mutable_iter iter;
+	XLogRecPtr	local_flush = GetFlushRecPtr();
+
+	*write = InvalidXLogRecPtr;
+	*flush = InvalidXLogRecPtr;
+
+	dlist_foreach_modify(iter, &lsn_mapping)
+	{
+		PGLFlushPosition *pos =
+			dlist_container(PGLFlushPosition, node, iter.cur);
+
+		*write = pos->remote_end;
+
+		if (pos->local_end <= local_flush)
+		{
+			*flush = pos->remote_end;
+			dlist_delete(iter.cur);
+			pfree(pos);
+		}
+		else
+		{
+			/*
+			 * Don't want to uselessly iterate over the rest of the list which
+			 * could potentially be long. Instead get the last element and
+			 * grab the write position from there.
+			 */
+			pos = dlist_tail_element(PGLFlushPosition, node,
+									 &lsn_mapping);
+			*write = pos->remote_end;
+			return false;
+		}
+	}
+
+	return dlist_is_empty(&lsn_mapping);
+}
+
 /*
  * Send a Standby Status Update message to server.
  *
@@ -1054,7 +1126,14 @@ send_feedback(PGconn *conn, XLogRecPtr recvpos, int64 now, bool force)
 	if (recvpos < last_recvpos)
 		recvpos = last_recvpos;
 
-	flushpos = writepos = recvpos;
+	if (get_flush_position(&writepos, &flushpos))
+	{
+		/*
+		 * No outstanding transactions to flush, we can report the latest
+		 * received position. This is important for synchronous replication.
+		 */
+		flushpos = writepos = recvpos;
+	}
 
 	if (writepos < last_writepos)
 		writepos = last_writepos;
@@ -1554,6 +1633,20 @@ pglogical_apply_main(Datum main_arg)
 	/* Connect to our database. */
 	BackgroundWorkerInitializeConnectionByOid(MyPGLogicalWorker->dboid, InvalidOid);
 
+	/* setup synchronous commit according to the user's wishes */
+	SetConfigOption("synchronous_commit",
+					pglogical_synchronous_commit ? "local" : "off",
+					PGC_BACKEND, PGC_S_OVERRIDE);	/* other context? */
+
+	/*
+	 * Disable function body checks during replay. That's necessary because a)
+	 * the creator of the function might have had it disabled b) the function
+	 * might be search_path dependant and we don't fix the contents of
+	 * functions.
+	 */
+	SetConfigOption("check_function_bodies", "off",
+					PGC_INTERNAL, PGC_S_OVERRIDE);
+
 	/* Load the subscription. */
 	StartTransactionCommand();
 	saved_ctx = MemoryContextSwitchTo(TopMemoryContext);
