Update multimaster

Author: knizhnik

contrib/multimaster/multimaster.c

@@ -1,7 +1,7 @@
 /*
- * pg_dtm.c
+ * multimaster.c
  *
- * Pluggable distributed transaction manager
+ * Multimaster based on logical replication
  *
  */
 

contrib/multimaster/receiver_raw.c

@@ -14,11 +14,13 @@
  */
 
 /* Some general headers for custom bgworker facility */
+#include <unistd.h>
 #include "postgres.h"
 #include "fmgr.h"
 #include "libpq-fe.h"
 #include "pqexpbuffer.h"
 #include "access/xact.h"
+#include "access/transam.h"
 #include "lib/stringinfo.h"
 #include "pgstat.h"
 #include "executor/spi.h"
@@ -50,7 +52,7 @@ static int receiver_idle_time = 1;
 static bool receiver_sync_mode = true;
 
 /* Worker name */
-static char *worker_name = "multimaster";
+char worker_proc[16];
 
 /* Lastly written positions */
 static XLogRecPtr output_written_lsn = InvalidXLogRecPtr;
@@ -93,7 +95,7 @@ sendFeedback(PGconn *conn, int64 now)
 	ereport(LOG, (errmsg("%s: confirming write up to %X/%X, "
 						 "flush to %X/%X (slot custom_slot), "
 						 "applied to %X/%X",
-						 worker_name,
+						 worker_proc,
 						 (uint32) (output_written_lsn >> 32),
 						 (uint32) output_written_lsn,
 						 (uint32) (output_fsync_lsn >> 32),
@@ -119,7 +121,7 @@ sendFeedback(PGconn *conn, int64 now)
 	if (PQputCopyData(conn, replybuf, len) <= 0 || PQflush(conn))
 	{
 		ereport(LOG, (errmsg("%s: could not send feedback packet: %s",
-							 worker_name, PQerrorMessage(conn))));
+							 worker_proc, PQerrorMessage(conn))));
 		return false;
 	}
 
@@ -209,13 +211,15 @@ receiver_raw_main(Datum main_arg)
 	PQExpBuffer query;
 	PGconn *conn;
 	PGresult *res;
+    TransactionId xid = InvalidTransactionId;
     bool insideTrans = false;
     bool rollbackTransaction = false;
 
 	/* Register functions for SIGTERM/SIGHUP management */
 	pqsignal(SIGHUP, receiver_raw_sighup);
 	pqsignal(SIGTERM, receiver_raw_sigterm);
 
+    sprintf(worker_proc, "mm_recv_%d", getpid());
 
 	/* We're now ready to receive signals */
 	BackgroundWorkerUnblockSignals();
@@ -229,13 +233,13 @@ receiver_raw_main(Datum main_arg)
 	{
 		PQfinish(conn);
 		ereport(ERROR, (errmsg("%s: Could not establish connection to remote server",
-							 worker_name)));
+							 worker_proc)));
 		proc_exit(1);
 	}
 
 	query = createPQExpBuffer();
 
-    appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" LOGICAL \"%s\"", args->receiver_slot, worker_name);
+    appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" LOGICAL \"%s\"", args->receiver_slot, worker_proc);
     res = PQexec(conn, query->data);
 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 	{
@@ -244,7 +248,7 @@ receiver_raw_main(Datum main_arg)
 		{
             PQclear(res);
             ereport(ERROR, (errmsg("%s: Could not create logical slot",
-                                 worker_name)));
+                                 worker_proc)));
             proc_exit(1);
         }
     }
@@ -259,7 +263,7 @@ receiver_raw_main(Datum main_arg)
 	{
 		PQclear(res);
 		ereport(LOG, (errmsg("%s: Could not start logical replication",
-							 worker_name)));
+							 worker_proc)));
 		proc_exit(1);
 	}
 	PQclear(res);
@@ -284,13 +288,13 @@ receiver_raw_main(Datum main_arg)
 			/* Process config file */
 			ProcessConfigFile(PGC_SIGHUP);
 			got_sighup = false;
-			ereport(LOG, (errmsg("%s: processed SIGHUP", worker_name)));
+			ereport(LOG, (errmsg("%s: processed SIGHUP", worker_proc)));
 		}
 
 		if (got_sigterm)
 		{
 			/* Simply exit */
-			ereport(LOG, (errmsg("%s: processed SIGTERM", worker_name)));
+			ereport(LOG, (errmsg("%s: processed SIGTERM", worker_proc)));
 			proc_exit(0);
 		}
 
@@ -342,15 +346,15 @@ receiver_raw_main(Datum main_arg)
 				walEnd = fe_recvint64(&copybuf[pos]);
 				ereport(LOG, (errmsg("%s: keepalive message from server, "
 									 "walEnd %X/%X, ",
-									 worker_name,
+									 worker_proc,
 									 (uint32) (walEnd >> 32),
 									 (uint32) walEnd)));
 				pos += 8;	/* read walEnd */
 				pos += 8;	/* skip sendTime */
 				if (rc < pos + 1)
 				{
 					ereport(LOG, (errmsg("%s: streaming header too small: %d",
-										 worker_name, rc)));
+										 worker_proc, rc)));
 					proc_exit(1);
 				}
 				replyRequested = copybuf[pos];
@@ -378,7 +382,7 @@ receiver_raw_main(Datum main_arg)
 			else if (copybuf[0] != 'w')
 			{
 				ereport(LOG, (errmsg("%s: Incorrect streaming header",
-									 worker_name)));
+									 worker_proc)));
 				proc_exit(1);
 			}
 
@@ -392,14 +396,14 @@ receiver_raw_main(Datum main_arg)
 			if (rc < hdr_len + 1)
 			{
 				ereport(LOG, (errmsg("%s: Streaming header too small",
-									 worker_name)));
+									 worker_proc)));
 				proc_exit(1);
 			}
 
 			/* Log some useful information */
 			ereport(LOG, (errmsg("%s: received from server, walStart %X/%X, "
 								 "and walEnd %X/%X",
-						 worker_name,
+						 worker_proc,
 						 (uint32) (walStart >> 32),
 						 (uint32) walStart,
 						 (uint32) (walEnd >> 32),
@@ -411,17 +415,13 @@ receiver_raw_main(Datum main_arg)
 			SetCurrentStatementStartTimestamp();
 
             if (strncmp(stmt, "BEGIN ", 6) == 0) { 
-                TransactionId xid;
                 int rc = sscanf(stmt + 6, "%u", &xid);
                 Assert(rc == 1);
                 Assert(!insideTrans);
                 SetCurrentStatementStartTimestamp();
                 MMJoinTransaction(xid);
 
                 StartTransactionCommand();
-                BeginTransactionBlock();
-                CommitTransactionCommand();
-
                 SPI_connect();
                 PushActiveSnapshot(GetTransactionSnapshot());
                 insideTrans = true;
@@ -431,19 +431,20 @@ receiver_raw_main(Datum main_arg)
                 insideTrans = false;
                 SPI_finish();
                 PopActiveSnapshot();
-                StartTransactionCommand();
                 if (rollbackTransaction) {
-                    UserAbortTransactionBlock();
-                } 
-                PG_TRY();
-                {
-                    CommitTransactionCommand();
-                }
-                PG_CATCH();
-                {
-                    elog(WARNING, "%s: Current transaction is aborted at receiver", worker_name);
+                    elog(WARNING, "%s: Rollback transaction %u", worker_proc, xid);
+                    AbortCurrentTransaction();
+                } else { 
+                    PG_TRY();
+                    {
+                        CommitTransactionCommand();
+                    }
+                    PG_CATCH();
+                    {
+                        elog(WARNING, "%s: Commit of transaction %u is failed", worker_proc, xid);
+                    }
+                    PG_END_TRY();
                 }
-                PG_END_TRY();
             } else if (!rollbackTransaction) {
                 Assert(insideTrans);
                 /* Execute query */
@@ -452,20 +453,20 @@ receiver_raw_main(Datum main_arg)
                     rc = SPI_execute(stmt, false, 0);
                     if (rc == SPI_OK_INSERT)
                         ereport(LOG, (errmsg("%s: INSERT received correctly: %s",
-                                             worker_name, stmt)));
+                                             worker_proc, stmt)));
                     else if (rc == SPI_OK_UPDATE)
                         ereport(LOG, (errmsg("%s: UPDATE received correctly: %s",
-                                             worker_name, stmt)));
+                                             worker_proc, stmt)));
                     else if (rc == SPI_OK_DELETE)
                         ereport(LOG, (errmsg("%s: DELETE received correctly: %s",
-                                             worker_name, stmt)));
+                                             worker_proc, stmt)));
                     else
                         ereport(WARNING, (errmsg("%s: Error when applying change: %s",
-                                             worker_name, stmt)));
+                                             worker_proc, stmt)));
                 }
                 PG_CATCH();
                 {
-                    elog(WARNING, "%s: %s failed at receiver", worker_name, stmt);
+                    elog(WARNING, "%s: %s failed in transaction %u", worker_proc, stmt, xid);
                     rollbackTransaction = true;
                 }
                 PG_END_TRY();
@@ -531,15 +532,15 @@ receiver_raw_main(Datum main_arg)
 			else if (r < 0)
 			{
 				ereport(LOG, (errmsg("%s: Incorrect status received... Leaving.",
-									 worker_name)));
+									 worker_proc)));
 				proc_exit(1);
 			}
 
 			/* Else there is actually data on the socket */
 			if (PQconsumeInput(conn) == 0)
 			{
 				ereport(LOG, (errmsg("%s: Data remaining on the socket... Leaving.",
-									 worker_name)));
+									 worker_proc)));
 				proc_exit(1);
 			}
 			continue;
@@ -549,15 +550,15 @@ receiver_raw_main(Datum main_arg)
 		if (rc == -1)
 		{
 			ereport(LOG, (errmsg("%s: COPY Stream has abruptly ended...",
-								 worker_name)));
+								 worker_proc)));
 			break;
 		}
 
 		/* Failure when reading copy stream, leave */
 		if (rc == -2)
 		{
 			ereport(LOG, (errmsg("%s: Failure while receiving changes...",
-								 worker_name)));
+								 worker_proc)));
 			proc_exit(1);
 		}
 	}
@@ -608,4 +609,9 @@ int MMStartReceivers(char* conns, int node_id)
             worker.bgw_main_arg = (Datum)ctx;
             RegisterBackgroundWorker(&worker);
         }
-        con
+        conn_str = p + 1;
+    }
+
+    return i;
+}
+

contrib/multimaster/tests/dtmbench

@@ -0,0 +1,93 @@
+# PostgreSQL Client Authentication Configuration File
+# ===================================================
+#
+# Refer to the "Client Authentication" section in the PostgreSQL
+# documentation for a complete description of this file.  A short
+# synopsis follows.
+#
+# This file controls: which hosts are allowed to connect, how clients
+# are authenticated, which PostgreSQL user names they can use, which
+# databases they can access.  Records take one of these forms:
+#
+# local      DATABASE  USER  METHOD  [OPTIONS]
+# host       DATABASE  USER  ADDRESS  METHOD  [OPTIONS]
+# hostssl    DATABASE  USER  ADDRESS  METHOD  [OPTIONS]
+# hostnossl  DATABASE  USER  ADDRESS  METHOD  [OPTIONS]
+#
+# (The uppercase items must be replaced by actual values.)
+#
+# The first field is the connection type: "local" is a Unix-domain
+# socket, "host" is either a plain or SSL-encrypted TCP/IP socket,
+# "hostssl" is an SSL-encrypted TCP/IP socket, and "hostnossl" is a
+# plain TCP/IP socket.
+#
+# DATABASE can be "all", "sameuser", "samerole", "replication", a
+# database name, or a comma-separated list thereof. The "all"
+# keyword does not match "replication". Access to replication
+# must be enabled in a separate record (see example below).
+#
+# USER can be "all", a user name, a group name prefixed with "+", or a
+# comma-separated list thereof.  In both the DATABASE and USER fields
+# you can also write a file name prefixed with "@" to include names
+# from a separate file.
+#
+# ADDRESS specifies the set of hosts the record matches.  It can be a
+# host name, or it is made up of an IP address and a CIDR mask that is
+# an integer (between 0 and 32 (IPv4) or 128 (IPv6) inclusive) that
+# specifies the number of significant bits in the mask.  A host name
+# that starts with a dot (.) matches a suffix of the actual host name.
+# Alternatively, you can write an IP address and netmask in separate
+# columns to specify the set of hosts.  Instead of a CIDR-address, you
+# can write "samehost" to match any of the server's own IP addresses,
+# or "samenet" to match any address in any subnet that the server is
+# directly connected to.
+#
+# METHOD can be "trust", "reject", "md5", "password", "gss", "sspi",
+# "ident", "peer", "pam", "ldap", "radius" or "cert".  Note that
+# "password" sends passwords in clear text; "md5" is preferred since
+# it sends encrypted passwords.
+#
+# OPTIONS are a set of options for the authentication in the format
+# NAME=VALUE.  The available options depend on the different
+# authentication methods -- refer to the "Client Authentication"
+# section in the documentation for a list of which options are
+# available for which authentication methods.
+#
+# Database and user names containing spaces, commas, quotes and other
+# special characters must be quoted.  Quoting one of the keywords
+# "all", "sameuser", "samerole" or "replication" makes the name lose
+# its special character, and just match a database or username with
+# that name.
+#
+# This file is read on server startup and when the postmaster receives
+# a SIGHUP signal.  If you edit the file on a running system, you have
+# to SIGHUP the postmaster for the changes to take effect.  You can
+# use "pg_ctl reload" to do that.
+
+# Put your actual configuration here
+# ----------------------------------
+#
+# If you want to allow non-local connections, you need to add more
+# "host" records.  In that case you will also need to make PostgreSQL
+# listen on a non-local interface via the listen_addresses
+# configuration parameter, or via the -i or -h command line switches.
+
+# CAUTION: Configuring the system for local "trust" authentication
+# allows any local user to connect as any PostgreSQL user, including
+# the database superuser.  If you do not trust all your local users,
+# use another authentication method.
+
+
+# TYPE  DATABASE        USER            ADDRESS                 METHOD
+
+# "local" is for Unix domain socket connections only
+local   all             all                                     trust
+# IPv4 local connections:
+host    all             all             127.0.0.1/32            trust
+# IPv6 local connections:
+host    all             all             ::1/128                 trust
+# Allow replication connections from localhost, by a user with the
+# replication privilege.
+local   replication     knizhnik                                trust
+host    replication     knizhnik        127.0.0.1/32            trust
+#host    replication     knizhnik        ::1/128                 trust

contrib/multimaster/tests/pg_hba.conf

@@ -547,9 +547,9 @@
 # These settings are initialized by initdb, but they can be changed.
 lc_messages = 'en_US.UTF-8'			# locale for system error message
 					# strings
-lc_monetary = 'ru_RU.UTF-8'			# locale for monetary formatting
-lc_numeric = 'ru_RU.UTF-8'			# locale for number formatting
-lc_time = 'ru_RU.UTF-8'				# locale for time formatting
+lc_monetary = 'en_US.UTF-8'			# locale for monetary formatting
+lc_numeric = 'en_US.UTF-8'			# locale for number formatting
+lc_time = 'en_US.UTF-8'				# locale for time formatting
 
 # default configuration for text search
 default_text_search_config = 'pg_catalog.english'