Skeleton for move_mpart

Author: arssher

Makefile

@@ -3,7 +3,7 @@ EXTENSION = pg_shardman        # the extension name
 DATA = pg_shardman--0.0.1.sql
 
 MODULE_big = pg_shardman
-OBJS = src/pg_shardman.o src/udf.o src/shard.o
+OBJS = src/pg_shardman.o src/udf.o src/shard.o src/timeutils.o
 
 PG_CPPFLAGS += -Isrc/include
 

pg_shardman--0.0.1.sql

@@ -111,7 +111,6 @@ DECLARE
 	fdw_part_name text;
 BEGIN
 	IF NEW.owner != (SELECT shardman.get_node_id()) THEN
-		raise info 'creating foreign table';
 		SELECT nodes.connstring FROM shardman.nodes WHERE id = NEW.owner
 		  INTO connstring;
 		EXECUTE format('DROP SERVER IF EXISTS %I CASCADE;', NEW.part_name);
@@ -208,7 +207,8 @@ CREATE TABLE local_meta (
 INSERT INTO @extschema@.local_meta VALUES ('node_id', NULL);
 
 -- available commands
-CREATE TYPE cmd AS ENUM ('add_node', 'rm_node', 'create_hash_partitions');
+CREATE TYPE cmd AS ENUM ('add_node', 'rm_node', 'create_hash_partitions',
+						 'move_mpart');
 -- command status
 CREATE TYPE cmd_status AS ENUM ('waiting', 'canceled', 'failed', 'in progress', 'success');
 
@@ -466,6 +466,21 @@ BEGIN
 	INSERT INTO @extschema@.cmd_opts VALUES (DEFAULT, c_id, partitions_count);
 END
 $$ LANGUAGE plpgsql;
+
+-- Move master partition to another node. Params:
+-- 'part_name' is name of the partition to move
+-- 'dest' is id of the destination node
+CREATE FUNCTION move_mpart(part_name text, dest int) RETURNS int AS $$
+DECLARE
+	c_id int;
+BEGIN
+	INSERT INTO @extschema@.cmd_log VALUES (DEFAULT, 'move_mpart')
+										   RETURNING id INTO c_id;
+	INSERT INTO @extschema@.cmd_opts VALUES (DEFAULT, c_id, part_name);
+	INSERT INTO @extschema@.cmd_opts VALUES (DEFAULT, c_id, dest);
+	RETURN c_id;
+END $$ LANGUAGE plpgsql;
+
 -- Otherwise partitioned tables on worker nodes not will be dropped properly,
 -- see pathman's docs.
 ALTER EVENT TRIGGER pathman_ddl_trigger ENABLE ALWAYS;

postgresql.conf.common.template

@@ -1,6 +1,9 @@
 shared_preload_libraries = 'pg_pathman, pg_shardman'
 
-log_min_messages = INFO
+# just to suppress logging
+autovacuum = off
+
+log_min_messages = DEBUG1
 # client_min_messages = NOTICE
 client_min_messages = INFO
 

src/include/pg_shardman.h

@@ -42,5 +42,6 @@ extern uint64 void_spi(char *sql);
 extern void update_cmd_status(int64 id, const char *new_status);
 extern void cmd_canceled(Cmd *cmd);
 extern char *get_worker_node_connstring(int node_id);
+extern int32 get_partition_owner(const char *part_name);
 
 #endif							/* PG_SHARDMAN_H */

src/include/timeutils.h

@@ -0,0 +1,10 @@
+#ifndef TIMEUTILS_H
+#define TIMEUTILS_H
+
+#include <time.h>
+
+extern int timespeccmp(struct timespec t1, struct timespec t2);
+extern struct timespec timespec_add_millis(struct timespec t, long millis);
+extern int timespec_diff_millis(struct timespec t1, struct timespec t2);
+
+#endif							/* TIMEUTILS_H */

src/pg_shardman.c

@@ -714,3 +714,39 @@ get_worker_node_connstring(int node_id)
 	SPI_EPILOG;
 	return res;
 }
+
+/*
+ * Get node id on which given partition is stored. -1 is returned if there is
+ * no such partition.
+ */
+int32
+get_partition_owner(const char *part_name)
+{
+	char *sql;
+	bool isnull;
+	int owner;
+
+	SPI_PROLOG;
+	sql = psprintf( /* allocated in SPI ctxt, freed with ctxt release */
+		"select owner from shardman.partitions where part_name = '%s';",
+		part_name);
+
+	if (SPI_execute(sql, true, 0) < 0)
+	{
+		shmn_elog(FATAL, "Stmt failed : %s", sql);
+	}
+
+	if (SPI_processed == 0)
+	{
+		owner = -1;
+	}
+	else
+	{
+		owner =	DatumGetInt32(SPI_getbinval(SPI_tuptable->vals[0],
+											SPI_tuptable->tupdesc,
+											1, &isnull));
+	}
+
+	SPI_EPILOG;
+	return owner;
+}

src/shard.c

@@ -7,8 +7,56 @@
  */
 #include "postgres.h"
 #include "libpq-fe.h"
+#include "lib/ilist.h"
+
+#include <time.h>
+#include <limits.h>
+#include <sys/epoll.h>
 
 #include "shard.h"
+#include "timeutils.h"
+
+typedef enum
+{
+	MOVEMPART_IN_PROGRESS,
+	MOVEMPART_FAILED,
+	MOVEMPART_SUCCESS
+} MoveMPartResult;
+
+/* result of one iteration of processing */
+typedef enum
+{
+	EXECMOVEMPART_EPOLL, /* add me to epoll on epolled_fd on EPOLLIN */
+	EXECMOVEMPART_WAKEMEUP, /* wake me up again on waketm */
+	EXECMOVEMPART_DONE /* the work is done, never invoke me again */
+} ExecMoveMPartRes;
+
+typedef struct
+{
+	const char *part_name; /* partition name */
+	int32 src_node; /* node we are moving partition from */
+	int32 dst_node; /* node we are moving partition to */
+	char *src_connstr;
+	char *dst_connstr;
+	struct timespec waketm; /* wake me up at waketm to do the job */
+	/* We need to epoll only on socket with dst to wait for copy */
+	 /* exec_move_mpart sets fd here when it wants to be wakened by epoll */
+	int fd_to_epoll;
+	int fd_in_epoll_set; /* socket *currently* in epoll set. -1 of none */
+	MoveMPartResult result;
+} MoveMPartState;
+
+typedef struct
+{
+	slist_node list_node;
+	MoveMPartState *mmps;
+} MoveMPartStateNode;
+
+static void init_mmp_state(MoveMPartState *mmps, const char *part_name,
+						   int32 dst_node);
+static void move_mparts(MoveMPartState *mmpss, int nparts);
+static int calc_timeout(slist_head *timeout_states);
+static ExecMoveMPartRes exec_move_mpart(MoveMPartState *mmps);
 
 /*
  * Steps are:
@@ -124,13 +172,14 @@ create_hash_partitions(Cmd *cmd)
 	cmd_canceled(cmd);
 }
 
-
 /*
  * Move master partition to specified node. We
  * - Disable subscription on destination, otherwise we can't drop rep slot on
      source.
  * - Idempotently create publication and repl slot on source.
- * - Idempotently create table and subscription on destination.
+ * - Idempotently create table and async subscription on destination.
+ *   We use async subscription, because sync would block table while copy is
+ *   in progress. But with async, we have to lock the table after initial sync.
  * - Now inital copy has started, remember that at least in ram to retry
  *   from this point if network fails.
  * - Sleep & check in connection to the dest waiting for completion of the
@@ -145,9 +194,150 @@ create_hash_partitions(Cmd *cmd)
  *  If we don't save progress (whether initial sync started or done, lsn,
  *  etc), we have to start everything from the ground if master reboots. This
  *  is arguably fine.
+ *
  */
 void
 move_mpart(Cmd *cmd)
+{
+	char *part_name = cmd->opts[0];
+	int32 dst_node = atoi(cmd->opts[1]);
+
+	MoveMPartState *mmps = palloc(sizeof(MoveMPartState));
+	init_mmp_state(mmps, part_name, dst_node);
+
+	move_mparts(mmps, 1);
+	update_cmd_status(cmd->id, "success");
+}
+
+
+/*
+ * Fill MoveMPartState, retrieving needed data. If something goes wrong, we
+ * don't bother to fill the rest of fields.
+ */
+void
+init_mmp_state(MoveMPartState *mmps, const char *part_name, int32 dst_node)
+{
+	int e;
+
+	mmps->part_name = part_name;
+	if ((mmps->src_node = get_partition_owner(part_name)) == -1)
+	{
+		shmn_elog(WARNING, "Partition %s doesn't exist, not moving it",
+				  part_name);
+		mmps->result = MOVEMPART_FAILED;
+		return;
+	}
+	mmps->dst_node = dst_node;
+
+	/* src_connstr is surely not NULL since src_node is referenced by
+	   part_name */
+	mmps->src_connstr = get_worker_node_connstr(mmps->src_node);
+	mmps->dst_connstr = get_worker_node_connstr(mmps->dst_node);
+	if (mmps->dst_connstr == NULL)
+	{
+		shmn_elog(WARNING, "Node %d doesn't exist, not moving %s to it",
+				  mmps->dst_node, part_name);
+		mmps->result = MOVEMPART_FAILED;
+		return;
+	}
+
+	/* Task is ready to be processed right now */
+	if ((e = clock_gettime(CLOCK_MONOTONIC, &mmps->waketm)) == -1)
+	{
+		shmn_elog(FATAL, "clock_gettime failed, %s", strerror(e));
+	}
+	mmps->fd_in_epoll_set = -1;
+
+	mmps->result = MOVEMPART_IN_PROGRESS;
+}
+
+/*
+ * Move partitions as specified in move_mpart_states list
+ */
+void
+move_mparts(MoveMPartState *mmpss, int nparts)
+{
+	/* list of sleeping mmp states we need to wake after specified timeout */
+	slist_head timeout_states = SLIST_STATIC_INIT(timeout_states);
+	slist_iter iter;
+
+	int timeout; /* at least one task will be ready after timeout millis */
+	int unfinished_moves = 0; /* number of not yet failed or succeeded tasks */
+	int i;
+	int e;
+	int epfd;
+
+	for (i = 0; i < nparts; i++)
+	{
+		if (mmpss[i].result != MOVEMPART_FAILED)
+		{
+			/* In the beginning, all tasks are ready immediately */
+			MoveMPartStateNode *mmps_node = palloc(sizeof(MoveMPartStateNode));
+			elog(DEBUG4, "Adding task %s to timeout list", mmpss[i].part_name);
+			mmps_node->mmps = &mmpss[i];
+			slist_push_head(&timeout_states, &mmps_node->list_node);
+			unfinished_moves++;
+		}
+	}
+
+	if ((epfd = epoll_create1(0)) == -1)
+	{
+		shmn_elog(FATAL, "epoll_create1 failed");
+	}
+
+	while (unfinished_moves > 0)
+	{
+		timeout = calc_timeout(&timeout_states);
+		unfinished_moves--;
+	}
+}
+
+/* Calculate when we need to wake if no epoll events are happening */
+int
+calc_timeout(slist_head *timeout_states)
+{
+	slist_iter iter;
+	struct timespec curtm;
+	int e;
+	int timeout = -1; /* If no tasks wait for us, don't wake */
+
+	slist_foreach(iter, timeout_states)
+	{
+		MoveMPartStateNode *mmps_node =
+			slist_container(MoveMPartStateNode, list_node, iter.cur);
+		MoveMPartState *mmps = mmps_node->mmps;
+		shmn_elog(DEBUG1, "Peeking into %s task wake time", mmps->part_name);
+		if ((e = clock_gettime(CLOCK_MONOTONIC, &curtm)) == -1)
+		{
+			shmn_elog(FATAL, "clock_gettime failed, %s", strerror(e));
+		}
+		if (timespeccmp(curtm, mmps->waketm) >= 0)
+		{
+			shmn_elog(DEBUG1, "Task %s is already ready", mmps->part_name);
+			timeout = 0;
+			return timeout;
+		}
+		else
+		{
+			int diff = Max(0, timespec_diff_millis(mmps->waketm, curtm));
+			if (timeout == -1)
+				timeout = diff;
+			else
+				timeout = Min(timeout, diff);
+			shmn_elog(DEBUG1, "Timeout set to %d due to task %s ",
+					  timeout, mmps->part_name);
+		}
+	}
+
+	return timeout;
+}
+
+/*
+ * Actually run MoveMPart state machine. Return value says when (if ever)
+ * we want to be executed again.
+ */
+ExecMoveMPartRes
+exec_move_mpart(MoveMPartState *mmps)
 {
 
 }

src/timeutils.c

@@ -0,0 +1,53 @@
+#include "timeutils.h"
+
+#define MILLION 1000000L
+#define BILLION 1000000000L
+
+/*
+ * Like strcmp, but for timespec. Returns -1 if t1 < t2.
+ */
+int
+timespeccmp(struct timespec t1, struct timespec t2)
+{
+	if (t1.tv_sec == t2.tv_sec)
+	{
+		if (t1.tv_nsec < t2.tv_nsec)
+			return -1;
+		else if (t1.tv_nsec > t2.tv_nsec)
+			return 1;
+		return 0;
+	}
+	else if (t1.tv_sec < t2.tv_sec)
+		return -1;
+	return 1;
+}
+
+/*
+ * Add milliseconds to timespec
+ */
+struct timespec
+timespec_add_millis(struct timespec t, long millis)
+{
+	time_t tv_sec;
+	long tv_nsec;
+
+	tv_sec = t.tv_sec + millis / 1000;
+	tv_nsec = t.tv_nsec + (millis % 1000) * MILLION;
+	if (tv_nsec >= BILLION) {
+		tv_nsec -= BILLION;
+		tv_sec++;
+	}
+
+	return (struct timespec) { .tv_sec = tv_sec, .tv_nsec = tv_nsec };
+}
+
+/*
+ * Get t1 - t2 difference in milliseconds. Not reliable if time_t is unsigned
+ */
+int
+timespec_diff_millis(struct timespec t1, struct timespec t2)
+{
+	int sec_diff = t1.tv_sec - t2.tv_sec;
+	long nsec_diff = t1.tv_nsec - t2.tv_nsec;
+	return sec_diff * 1000 + nsec_diff / 1000;
+}