node management functions: join_node and detect basebackups

Author: kelvich

multimaster--1.0.sql

@@ -24,7 +24,8 @@ LANGUAGE C;
 CREATE TABLE mtm.nodes(
     id int primary key not null,
     conninfo text not null,
-    is_self bool not null
+    is_self bool not null,
+    init_done bool not null default 'f'
 );
 
 CREATE FUNCTION mtm.after_node_create()
@@ -47,6 +48,11 @@ CREATE TRIGGER on_node_drop
     FOR EACH ROW
     EXECUTE FUNCTION mtm.after_node_drop();
 
+CREATE FUNCTION mtm.join_node(node_id int)
+RETURNS VOID
+AS 'MODULE_PATHNAME','mtm_join_node'
+LANGUAGE C;
+
 ---
 --- User facing API for node management.
 ---
@@ -57,6 +63,7 @@ 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;
+    -- XXX
     EXECUTE 'SET mtm.emerging_node_id = ' || node_id || ';';
     INSERT INTO mtm.nodes SELECT
         ordinality::int as id,
@@ -69,6 +76,27 @@ END
 $$
 LANGUAGE plpgsql;
 
+CREATE OR REPLACE FUNCTION mtm.add_node(connstr text) RETURNS void AS
+$$
+DECLARE
+    new_node_id int;
+BEGIN
+    -- XXX: add only to a configured mm?
+
+    INSERT INTO mtm.nodes SELECT
+        min(unused_ids.id), connstr, 'false'
+    FROM (
+        SELECT id FROM generate_series(1,16) id
+        EXCEPT
+        SELECT id FROM mtm.nodes
+    ) unused_ids
+    RETURNING id INTO new_node_id;
+
+    -- SELECT mtm.node_join(new_node_id, connstr);
+END
+$$
+LANGUAGE plpgsql;
+
 ---
 --- Various
 ---

src/bgwpool.c

@@ -199,6 +199,18 @@ BgwPoolStart(BgwPool* pool, char *poolName, Oid db_id, Oid user_id)
 	strncpy(pool->poolName, poolName, MAX_NAME_LEN);
 	pool->db_id = db_id;
 	pool->user_id = user_id;
+	pool->nWorkers = 0;
+	pool->shutdown = false;
+	pool->producerBlocked = false;
+	pool->head = 0;
+	pool->tail = 0;
+	pool->active = 0;
+	pool->pending = 0;
+	pool->lastPeakTime = 0;
+	pool->lastDynamicWorkerStartTime = 0;
+
+	PGSemaphoreReset(pool->available);
+	PGSemaphoreReset(pool->overflow);
 }
 
 size_t BgwPoolGetQueueSize(BgwPool* pool)

src/commit.c

@@ -248,7 +248,7 @@ MtmTwoPhaseCommit()
 	dmq_stream_unsubscribe(stream);
 	mtm_log(MtmTxTrace, "%s unsubscribed for %s", gid, stream);
 
-	MaybeLogSyncpoint();
+	MaybeLogSyncpoint(false);
 
 	return true;
 }

src/include/logger.h

@@ -75,6 +75,9 @@ typedef enum MtmLogTag
 	/* syncpoints */
 	SyncpointCreated		= LOG,
 	SyncpointApply			= LOG,
+
+	/* Node add/drop */
+	NodeMgmt				= LOG
 } MtmLogTag;
 
 // XXX: also meaningful process name would be cool

src/include/multimaster.h

@@ -56,11 +56,12 @@ typedef uint64 nodemask_t;
 /*
  * Definitions for the "mtm.nodes" table.
  */
-#define MTM_NODES					"nodes"
+#define MTM_NODES					"mtm.nodes"
 #define Natts_mtm_nodes				3
 #define Anum_mtm_nodes_id			1	/* node_id, same accross cluster */
 #define Anum_mtm_nodes_connifo		2	/* connection string */
 #define Anum_mtm_nodes_is_self		3	/* is that tuple for our node? */
+#define Anum_mtm_nodes_init_done	4	/* did monitor already create slots? */
 
 
 /* Identifier of global transaction */
@@ -145,6 +146,7 @@ typedef struct
 	int			node_id;
 	char	   *conninfo;
 	RepOriginId	origin_id;
+	bool		init_done;
 } MtmNode;
 
 typedef struct
@@ -213,6 +215,7 @@ extern bool MtmBackgroundWorker;
 extern bool MtmIsLogicalReceiver;
 extern bool MtmIsReceiver;
 extern bool MtmIsPoolWorker;
+extern bool MtmIsMonitorWorker;
 
 /* GUCs */
 extern int	MtmTransSpillThreshold;

src/include/syncpoint.h

@@ -33,7 +33,7 @@ typedef struct
 	XLogRecPtr	origin_lsn;
 } FilterEntry;
 
-extern void MaybeLogSyncpoint(void);
+extern void MaybeLogSyncpoint(bool force);
 extern void SyncpointRegister(int node_id, XLogRecPtr origin_lsn,
 							  XLogRecPtr local_lsn, XLogRecPtr trim_lsn);
 extern Syncpoint SyncpointGetLatest(int node_id);