Merge branch 'xtm' of https://gitlab.postgrespro.ru/pgpro-dev/postgrespro into xtm

Author: knizhnik

contrib/pg_dtm/README

@@ -43,6 +43,13 @@ The format of all commands:
 
 The commands:
 
+'h': hello()
+	The first message.
+
+	The arbiter replies with:
+		[RES_OK] if ready
+		[RES_FAILED] (or disconnection) if not ready
+
 'r': reserve(minxid, minsize)
 	Claims a sequence ≥ minsize of xids ≥ minxid for local usage. This will
 	prevent the arbiter from using those values for global transactions.
@@ -88,7 +95,7 @@ The commands:
 
 	The reply and 'wait' logic is the same as for the 'status' command.
 
-'h': snapshot(xid)
+'t': snapshot(xid)
 	Tells the arbiter to generate a snapshot for the global transaction
 	identified by the given 'xid'. The arbiter will create a snapshot for
 	every participant, so when each of them asks for the snapshot it will

contrib/pg_dtm/dtmd/Makefile

@@ -2,7 +2,6 @@ CC=gcc
 CFLAGS=-g -O2 -Wall -Iinclude -D_LARGEFILE64_SOURCE # -DDEBUG
 SOCKHUB_PREFIX=../sockhub
 SOCKHUB_CFLAGS=-I"$(SOCKHUB_PREFIX)"
-SOCKHUB_LDFLAGS=-lsockhub -L"$(SOCKHUB_PREFIX)"
 
 SYSTEM=$(shell uname -s)
 ifeq ($(SYSTEM),Darwin)
@@ -11,15 +10,20 @@ endif
 
 .PHONY: all clean check bindir objdir
 
-all: bin/dtmd
+all: bin/dtmd bin/heart
 	@echo Done.
 	@echo Feel free to run the tests with \'make check\'.
 
-bin/dtmd: obj/server.o obj/main.o obj/clog.o obj/clogfile.o obj/util.o obj/transaction.o obj/snapshot.o obj/ddd.o | bindir objdir
+bin/dtmd: obj/server.o obj/raft.o obj/main.o obj/clog.o obj/clogfile.o obj/util.o obj/transaction.o obj/snapshot.o obj/ddd.o | bindir objdir
 	$(CC) -o bin/dtmd $(CFLAGS) \
-		obj/server.o obj/main.o \
+		obj/server.o obj/raft.o obj/main.o \
 		obj/clog.o obj/clogfile.o obj/util.o obj/transaction.o \
-		obj/snapshot.o obj/ddd.o \
+		obj/snapshot.o obj/ddd.o
+
+bin/heart: obj/heart.o obj/raft.o obj/util.o | bindir objdir
+	$(CC) -o bin/heart $(CFLAGS) \
+		obj/heart.o obj/raft.o obj/util.o \
+		obj/snapshot.o \
 		$(SOCKHUB_LDFLAGS)
 
 obj/server.o: src/server.c | objdir

contrib/pg_dtm/dtmd/include/dtmdlimits.h

@@ -0,0 +1,23 @@
+#ifndef DTMD_LIMITS_H
+#define DTMD_LIMITS_H
+
+// how many xids are reserved per raft term
+#define XIDS_PER_TERM  1000000
+
+// start a new term when this number of xids is left
+#define NEW_TERM_THRESHOLD 100000
+
+#define MAX_TRANSACTIONS 4096
+
+#define BUFFER_SIZE (64 * 1024)
+#define LISTEN_QUEUE_SIZE 100
+#define MAX_STREAMS 4096
+
+#define MAX_SERVERS 16
+#define HEARTBEAT_TIMEOUT_MS 20
+#define ELECTION_TIMEOUT_MS_MIN 150
+#define ELECTION_TIMEOUT_MS_MAX 300
+#define RAFT_LOGLEN 1024
+#define RAFT_KEEP_APPLIED 512 // how many applied entries to keep during compaction
+
+#endif

contrib/pg_dtm/dtmd/include/limits.h

@@ -1,16 +1,18 @@
 #ifndef PROTO_H
 #define PROTO_H
 
+#define CMD_HELLO    'h'
 #define CMD_RESERVE  'r'
 #define CMD_BEGIN    'b'
 #define CMD_FOR      'y'
 #define CMD_AGAINST  'n'
-#define CMD_SNAPSHOT 'h'
+#define CMD_SNAPSHOT 't'
 #define CMD_STATUS   's'
 #define CMD_DEADLOCK 'd'
 
 #define RES_FAILED 0xDEADBEEF
 #define RES_OK 0xC0FFEE
+#define RES_REDIRECT 404
 #define RES_DEADLOCK 0xDEADDEED
 #define RES_TRANSACTION_COMMITTED 1
 #define RES_TRANSACTION_ABORTED 2

contrib/pg_dtm/dtmd/include/proto.h

@@ -0,0 +1,145 @@
+#ifndef RAFT_H
+#define RAFT_H
+
+#include <arpa/inet.h>
+#include <stdbool.h>
+#include "dtmdlimits.h"
+
+#define NOBODY -1
+
+#define MAJORITY_IS_NOT_ENOUGH // wait for unanimous ack for applying a new entry
+
+#define DEFAULT_LISTENHOST "0.0.0.0"
+#define DEFAULT_LISTENPORT 5431
+
+#define ROLE_FOLLOWER 0
+#define ROLE_CANDIDATE 1
+#define ROLE_LEADER 2
+
+#if RAFT_KEEP_APPLIED >= RAFT_LOGLEN
+#error please ensure RAFT_KEEP_APPLIED < RAFT_LOGLEN
+#endif
+
+#if HEARTBEAT_TIMEOUT_MS >= ELECTION_TIMEOUT_MS_MIN
+#error please ensure HEARTBEAT_TIMEOUT_MS < ELECTION_TIMEOUT_MS_MIN (considerably)
+#endif
+
+#if ELECTION_TIMEOUT_MS_MIN >= ELECTION_TIMEOUT_MS_MAX
+#error please ensure ELECTION_TIMEOUT_MS_MIN < ELECTION_TIMEOUT_MS_MAX
+#endif
+
+// raft module does not care what you mean by action and argument
+typedef struct raft_entry_t {
+	int term;
+	bool snapshot; // true if this is a snapshot entry
+	union {
+		struct { // snapshot == false
+			int action;
+			int argument;
+		};
+		struct { // snapshot == true
+			int minarg;
+			int maxarg;
+		};
+	};
+} raft_entry_t;
+
+typedef void (*raft_applier_t)(int action, int argument);
+
+typedef struct raft_log_t {
+	int first;
+	int size;    // number of entries past first
+	int acked;   // number of entries replicated to the majority of servers
+	int applied; // number of entries applied to the state machine
+	raft_entry_t entries[RAFT_LOGLEN]; // wraps around
+} raft_log_t;
+
+typedef struct raft_server_t {
+	int seqno;  // the rpc sequence number
+	int tosend; // index of the next entry to send
+	int acked;  // index of the highest entry known to be replicated
+
+	char *host;
+	int port;
+	struct sockaddr_in addr;
+} raft_server_t;
+
+typedef struct raft_t {
+	int term;   // current term (latest term we have seen)
+	int vote;   // who received our vote in current term
+	int role;
+	int me;     // my id
+	int votes;  // how many votes are for me (if candidate)
+	int leader; // the id of the leader
+	raft_log_t log;
+
+	int sock;
+
+	int servernum;
+	raft_server_t servers[MAX_SERVERS];
+
+	int timer;
+
+	raft_applier_t applier;
+} raft_t;
+
+#define RAFT_LOG(RAFT, INDEX) ((RAFT)->log.entries[(INDEX) % (RAFT_LOGLEN)])
+
+#define RAFT_MSG_UPDATE 0 // append entry
+#define RAFT_MSG_DONE   1 // entry appended
+#define RAFT_MSG_CLAIM  2 // vote for me
+#define RAFT_MSG_VOTE   3 // my vote
+
+typedef struct raft_msg_t {
+	int msgtype;
+	int term;
+	int from;
+	int seqno;
+} raft_msg_t;
+
+typedef struct raft_msg_update_t {
+	raft_msg_t msg;
+	int previndex; // the index of the preceding log entry
+	int prevterm;  // the term of the preceding log entry
+
+	bool empty;    // the message is just a heartbeat if empty
+	raft_entry_t entry;
+
+	int acked;     // the leader's acked number
+} raft_msg_update_t;
+
+typedef struct raft_msg_done_t {
+	raft_msg_t msg;
+	int index; // the index of the appended entry
+	int term;  // the term of the appended entry
+	bool success;
+} raft_msg_done_t;
+
+typedef struct raft_msg_claim_t {
+	raft_msg_t msg;
+	int index; // the index of my last entry
+	int term;  // the term of my last entry
+} raft_msg_claim_t;
+
+typedef struct raft_msg_vote_t {
+	raft_msg_t msg;
+	bool granted;
+} raft_msg_vote_t;
+
+// configuration
+void raft_init(raft_t *r);
+bool raft_add_server(raft_t *r, char *host, int port);
+bool raft_set_myid(raft_t *r, int myid);
+
+// log actions
+bool raft_emit(raft_t *r, int action, int argument);
+int raft_apply(raft_t *r, raft_applier_t applier);
+
+// control
+void raft_tick(raft_t *r, int msec);
+void raft_handle_message(raft_t *r, raft_msg_t *m);
+raft_msg_t *raft_recv_message(raft_t *r);
+int raft_create_udp_socket(raft_t *r);
+void raft_start_next_term(raft_t *r);
+
+#endif

contrib/pg_dtm/dtmd/include/raft.h

@@ -40,16 +40,37 @@ server_t server_init(
 	ondisconnect_callback_t ondisconnect
 );
 
+/*
+ * Assigns the given raft socket to the server. The server will add the socket
+ * to the 'select' calls and give you the incoming messages.
+ */
+void server_set_raft_socket(server_t server, int sock);
+
 /*
  * Starts the server. Returns 'true' on success, 'false' otherwise.
  */
 bool server_start(server_t server);
 
 /*
- * The main server loop. Does not return, so use the callbacks and signal
- * handlers to add more logic.
+ * The main server loop. Returns true if there is a raft message ready, or NULL
+ * if timed out. Use the callbacks and signal handlers to add more logic.
+ */
+bool server_tick(server_t server, int timeout_ms);
+
+/*
+ * Closes all client connections on the server and refuses to accept new ones.
+ */
+void server_disable(server_t server);
+
+/*
+ * Allows the server to accept new connections.
  */
-void server_loop(server_t server);
+void server_enable(server_t server);
+
+/*
+ * Enables or disables the server depending on the argument.
+ */
+void server_set_enabled(server_t server, bool enable);
 
 /*
  * These two methods allow you to set and get your custom 'userdata' for the
@@ -98,6 +119,13 @@ bool client_message_finish(client_t client);
  */
 bool client_message_shortcut(client_t client, xid_t arg);
 
+/*
+ * A shortcut to send the 'redirect' message.
+ *
+ * Returns 'true' on success, 'false' otherwise.
+ */
+bool client_redirect(client_t client, unsigned addr, int port);
+
 unsigned client_get_ip_addr(client_t client);
 
 #endif

contrib/pg_dtm/dtmd/include/server.h

@@ -2,7 +2,7 @@
 #define SNAPSHOT_H
 
 #include "int.h"
-#include "limits.h"
+#include "dtmdlimits.h"
 
 typedef struct Snapshot {
 	xid_t xmin;

contrib/pg_dtm/dtmd/include/snapshot.h

@@ -5,7 +5,7 @@
 #include "int.h"
 #include "clog.h"
 #include "snapshot.h"
-#include "limits.h"
+#include "dtmdlimits.h"
 
 #define MAX_SNAPSHOTS_PER_TRANS 8
 
@@ -68,6 +68,6 @@ int transaction_status(Transaction *t);
 void transaction_clear(Transaction *t);
 void transaction_push_listener(Transaction *t, char cmd, void *listener);
 void *transaction_pop_listener(Transaction *t, char cmd);
-bool transaction_participate(Transaction *t, int clientid);
+bool transaction_remove_listener(Transaction *t, char cmd, void *listener);
 
 #endif

contrib/pg_dtm/dtmd/include/transaction.h

@@ -7,15 +7,40 @@
 
 #include <stdbool.h>
 #include <sys/stat.h>
+#include <sys/time.h>
 #include <fcntl.h>
 #include <stdio.h>
+#include <stdlib.h>
 
 #include "int.h"
 
 char *join_path(const char *dir, const char *file);
 bool inrange(xid_t min, xid_t x, xid_t max);
 int falloc(int fd, off64_t size);
 
+static inline int min(int a, int b) {
+	return a < b ? a : b;
+}
+
+static inline int max(int a, int b) {
+	return a > b ? a : b;
+}
+
+static inline int rand_between(int min, int max) {
+	return rand() % (max - min + 1) + min;
+}
+
+// ------ timing ------
+
+typedef struct mstimer_t {
+	struct timeval tv;
+} mstimer_t;
+
+int mstimer_reset(mstimer_t *t);
+struct timeval ms2tv(int ms);
+
+// ------ logging ------
+
 #ifndef DEBUG
 #define debug(...)
 #else