node management functions: low-level add/drop functions, config reload in workers

Author: kelvich

Cluster.pm

@@ -176,10 +176,16 @@ sub start
 	foreach my $node (@$nodes)
 	{
 		$node->start();
+		note( "Starting node with connstr 'port=@{[ $node->port() ]} host=@{[ $node->host() ]}'");
+	}
+
+	$self->await_nodes( (0..$self->{nodenum}-1) );
+
+	foreach my $node (@$nodes)
+	{
 		$node->safe_psql($node->{dbname}, "create extension multimaster;");
 		$node->safe_psql($node->{dbname}, "select mtm.init_node($node_id, '{$connstrs}');");
 		$node_id = $node_id + 1;
-		note( "Starting node with connstr 'port=@{[ $node->port() ]} host=@{[ $node->host() ]}'");
 	}
 }
 

multimaster--1.0.sql

@@ -10,18 +10,71 @@ BEGIN
 END
 $$;
 
-
+-- message queue receiver, for internal use only
 CREATE FUNCTION mtm.dmq_receiver_loop(sender_name text) RETURNS void
 AS 'MODULE_PATHNAME','dmq_receiver_loop'
 LANGUAGE C;
 
+---
+--- Plumbering of node management: internal tables and triggers.
+--- Not indended to be used directly by users but rather through add/drop/init
+--- functions.
+---
 
-CREATE FUNCTION mtm.make_table_local(relation regclass) RETURNS void
-AS 'MODULE_PATHNAME','mtm_make_table_local'
+CREATE TABLE mtm.nodes(
+    id int primary key not null,
+    conninfo text not null,
+    is_self bool not null
+);
+
+CREATE FUNCTION mtm.after_node_create()
+RETURNS TRIGGER
+AS 'MODULE_PATHNAME','mtm_after_node_create'
 LANGUAGE C;
 
-CREATE FUNCTION mtm.init_node(node_id integer, connstrs text[]) RETURNS void
-AS 'MODULE_PATHNAME','mtm_init_node'
+CREATE TRIGGER on_node_create
+    AFTER INSERT ON mtm.nodes
+    FOR EACH ROW
+    EXECUTE FUNCTION mtm.after_node_create();
+
+CREATE FUNCTION mtm.after_node_drop()
+RETURNS TRIGGER
+AS 'MODULE_PATHNAME','mtm_after_node_drop'
+LANGUAGE C;
+
+CREATE TRIGGER on_node_drop
+    AFTER DELETE ON mtm.nodes
+    FOR EACH ROW
+    EXECUTE FUNCTION mtm.after_node_drop();
+
+---
+--- User facing API for node management.
+---
+
+CREATE OR REPLACE FUNCTION mtm.init_node(node_id integer, connstrs text[]) RETURNS void AS
+$$
+BEGIN
+    IF node_id <= 0 OR node_id > least(16, array_length(connstrs, 1)) THEN
+        RAISE EXCEPTION 'node_id should be in range [1 .. length(connstrs)]';
+    END IF;
+    EXECUTE 'SET mtm.emerging_node_id = ' || node_id || ';';
+    INSERT INTO mtm.nodes SELECT
+        ordinality::int as id,
+        unnest as conninfo,
+        ordinality = current_setting('mtm.emerging_node_id')::int as is_self
+    FROM
+        unnest(connstrs)
+    WITH ORDINALITY;
+END
+$$
+LANGUAGE plpgsql;
+
+---
+--- Various
+---
+
+CREATE FUNCTION mtm.make_table_local(relation regclass) RETURNS void
+AS 'MODULE_PATHNAME','mtm_make_table_local'
 LANGUAGE C;
 
 CREATE FUNCTION mtm.dump_lock_graph() RETURNS text
@@ -50,11 +103,6 @@ CREATE TABLE mtm.syncpoints(
     primary key(node_id, origin_lsn)
 );
 
-CREATE TABLE mtm.nodes(
-    id int primary key not null,
-    conninfo text not null,
-    is_self bool not null
-);
 
 CREATE OR REPLACE FUNCTION mtm.alter_sequences() RETURNS boolean AS
 $$

src/bgwpool.c

@@ -13,6 +13,8 @@
 #include "tcop/pquery.h"
 #include "utils/guc.h"
 #include "tcop/tcopprot.h"
+#include "utils/syscache.h"
+#include "utils/inval.h"
 
 #include "bgwpool.h"
 #include "multimaster.h"
@@ -40,6 +42,13 @@ BgwShutdownHandler(int sig)
 	die(sig);
 }
 
+static void
+subscription_change_cb(Datum arg, int cacheid, uint32 hashvalue)
+{
+	receiver_mtm_cfg_valid = false;
+}
+
+
 static void
 BgwPoolMainLoop(BgwPool* pool)
 {
@@ -70,6 +79,10 @@ BgwPoolMainLoop(BgwPool* pool)
 	ActivePortal->sourceText = "";
 
 	receiver_mtm_cfg = MtmLoadConfig();
+	/* Keep us informed about subscription changes. */
+	CacheRegisterSyscacheCallback(SUBSCRIPTIONOID,
+								  subscription_change_cb,
+								  (Datum) 0);
 
 	while (true)
 	{
@@ -209,7 +222,7 @@ static void BgwStartExtraWorker(BgwPool* pool)
 	MemSet(&worker, 0, sizeof(BackgroundWorker));
 	worker.bgw_flags = BGWORKER_SHMEM_ACCESS |  BGWORKER_BACKEND_DATABASE_CONNECTION;
 	worker.bgw_start_time = BgWorkerStart_ConsistentState;
-	worker.bgw_restart_time = MULTIMASTER_BGW_RESTART_TIMEOUT;
+	worker.bgw_restart_time = BGW_NEVER_RESTART;
 	worker.bgw_main_arg = PointerGetDatum(pool);
 	sprintf(worker.bgw_library_name, "multimaster");
 	sprintf(worker.bgw_function_name, "BgwPoolDynamicWorkerMainLoop");

src/commit.c

@@ -14,6 +14,8 @@
 #include "access/transam.h"
 #include "storage/proc.h"
 #include "utils/guc.h"
+#include "utils/syscache.h"
+#include "utils/inval.h"
 #include "miscadmin.h"
 #include "commands/dbcommands.h"
 #include "catalog/pg_subscription.h"
@@ -26,7 +28,9 @@
 #include "state.h"
 #include "syncpoint.h"
 
-static bool	backend_init_done;
+static bool	subchange_cb_registered;
+static bool	config_valid;
+// XXX: change dmq api and avoid that
 static int	sender_to_node[MTM_MAX_NODES];
 static MtmConfig *mtm_cfg;
 
@@ -38,22 +42,24 @@ static void GatherPrecommits(TransactionId xid, nodemask_t participantsMask,
 						     MtmMessageCode code);
 
 static void
-backend_init()
+pubsub_change_cb(Datum arg, int cacheid, uint32 hashvalue)
 {
-	int		i,
-			sender_id = 0;
+	config_valid = false;
+}
 
-	/* load mm config */
-	mtm_cfg = MtmLoadConfig();
+static void
+attach_node(int node_id, MtmConfig *new_cfg, Datum arg)
+{
+	int sender_id = dmq_attach_receiver(psprintf(MTM_DMQNAME_FMT, node_id),
+										node_id - 1);
+	sender_to_node[sender_id] = node_id;
+}
 
-	/* attach ourself to receiver queues and fill out sender_to_node[] */
-	for (i = 0; i < mtm_cfg->n_nodes; i++)
-	{
-		if (i + 1 == mtm_cfg->my_node_id)
-			continue;
-		dmq_attach_receiver(psprintf("node%d", i + 1), i);
-		sender_to_node[sender_id++] = i + 1;
-	}
+static void
+detach_node(int node_id, MtmConfig *new_cfg, Datum arg)
+{
+	/* detach incoming queues from this node */
+	dmq_detach_receiver(psprintf(MTM_DMQNAME_FMT, node_id));
 }
 
 void
@@ -103,11 +109,23 @@ MtmBeginTransaction()
 		return;
 	}
 
-	// XXX: reload on new node. (origin inval callback?)
-	if (!backend_init_done)
+	if (!subchange_cb_registered)
+	{
+		/* Keep us informed about subscription changes. */
+		CacheRegisterSyscacheCallback(SUBSCRIPTIONOID,
+								  pubsub_change_cb,
+								  (Datum) 0);
+		CacheRegisterSyscacheCallback(PUBLICATIONOID,
+								  pubsub_change_cb,
+								  (Datum) 0);
+		subchange_cb_registered = true;
+	}
+
+	AcceptInvalidationMessages();
+	if (!config_valid)
 	{
-		backend_init();
-		backend_init_done = true;
+		mtm_cfg = MtmReloadConfig(mtm_cfg, attach_node, detach_node, (Datum) NULL);
+		config_valid = true;
 	}
 
 	/* Reset MtmTx */
@@ -120,7 +138,6 @@ MtmBeginTransaction()
 
 	MtmDDLResetStatement();
 
-
 	/* Application name can be changed using PGAPPNAME environment variable */
 	if (Mtm->status != MTM_ONLINE
 		&& strcmp(application_name, MULTIMASTER_ADMIN) != 0
@@ -191,9 +208,8 @@ MtmTwoPhaseCommit()
 	LWLockAcquire(MtmCommitBarrier, LW_SHARED);
 
 	MtmLock(LW_SHARED);
-	participantsMask = (((nodemask_t)1 << mtm_cfg->n_nodes) - 1) &
-								  ~Mtm->disabledNodeMask &
-								  ~((nodemask_t)1 << (mtm_cfg->my_node_id-1));
+	participantsMask = ~Mtm->disabledNodeMask &
+						~((nodemask_t)1 << (mtm_cfg->my_node_id-1));
 	if (Mtm->status != MTM_ONLINE)
 		mtm_log(ERROR, "This node became offline during current transaction");
 	MtmUnlock();
@@ -242,8 +258,6 @@ GatherPrepares(TransactionId xid, nodemask_t participantsMask, int *failed_at)
 {
 	bool prepared = true;
 
-	Assert(participantsMask != 0);
-
 	while (participantsMask != 0)
 	{
 		bool ret;
@@ -311,8 +325,6 @@ GatherPrepares(TransactionId xid, nodemask_t participantsMask, int *failed_at)
 static void
 GatherPrecommits(TransactionId xid, nodemask_t participantsMask, MtmMessageCode code)
 {
-	Assert(participantsMask != 0);
-
 	while (participantsMask != 0)
 	{
 		bool ret;