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

Author: vbwagner

contrib/mmts/Cluster.pm

@@ -116,7 +116,7 @@ sub configure
 		my $unix_sock_dir = $ENV{PGHOST};
 
 		$node->append_conf("postgresql.conf", qq(
-			log_statement = none
+			log_statement = all
 			listen_addresses = '$host'
 			unix_socket_directories = '$unix_sock_dir'
 			port = $pgport
@@ -141,7 +141,7 @@ sub configure
 			multimaster.max_nodes = 6
 			multimaster.ignore_tables_without_pk = false
 			multimaster.queue_size = 4194304
-			log_line_prefix = '%t: '
+			log_line_prefix = '%t [%p]: '
 		));
 
 		$node->append_conf("pg_hba.conf", qq(

contrib/mmts/pglogical_apply.c

@@ -1067,8 +1067,8 @@ process_remote_update(StringInfo s, Relation rel)
 	bool		    found_tuple;
 	TupleData       old_tuple;
 	TupleData       new_tuple;
-	Oid			    idxoid;
-	Relation	    idxrel = InvalidOid;
+	Oid			    idxoid = InvalidOid;
+	Relation	    idxrel;
 	TupleDesc	    tupDesc = RelationGetDescr(rel);
 	ScanKeyData     skey[INDEX_MAX_KEYS];
 	HeapTuple	    remote_tuple = NULL;
@@ -1168,7 +1168,7 @@ process_remote_update(StringInfo s, Relation rel)
 	}
 
     /* release locks upon commit */
-	if (OidIsValid(idxrel))
+	if (OidIsValid(idxoid))
 		index_close(idxrel, NoLock);
 
 	PopActiveSnapshot();

contrib/mmts/pglogical_receiver.c

@@ -663,7 +663,8 @@ pglogical_receiver_main(Datum main_arg)
 				timeout.tv_usec = usecs;
 				timeoutptr = &timeout;
 
-				r = pg_select(PQsocket(conn) + 1, &input_mask, NULL, NULL, timeoutptr, conn->isRsocket);
+				r = PQselect(PQsocket(conn) + 1, &input_mask, NULL, NULL, timeoutptr,
+							 PQisRsocket(conn));
 				if (r == 0)
 				{
 					int64 now = feGetCurrentTimestamp();

contrib/mmts/state.c

@@ -247,9 +247,9 @@ MtmStateProcessNeighborEvent(int node_id, MtmNeighborEvent ev) // XXXX camelcase
 			 */
 			if (!BIT_CHECK(Mtm->disabledNodeMask, node_id-1))
 			{
-				BIT_SET(Mtm->disabledNodeMask, node_id-1);
+				MTM_LOG1("[WARN] node %d started recovery, but it wasn't disabled", node_id);
+				MtmOnNodeDisconnect(node_id-1);
 			}
-			// BIT_SET(Mtm->pglogicalSenderMask, node_id - 1);
 			break;
 
 		case MTM_NEIGHBOR_WAL_SENDER_START_RECOVERED:

src/common/pg_rsocket.c

@@ -49,10 +49,17 @@
 #define RDMACM_NAME "librdmacm.so.1"
 
 /* Declarations of rsocket functions */
-PgSocketCall *rcalls = NULL;
+static PgSocketCall *rcalls = NULL;
 
 static void *rdmacm_handle = NULL;
 
+PgSocketCall * pg_sock_vtab(void)
+{
+	if (!rcalls)
+		initialize_rsocket();
+	return rcalls;
+}
+
 /*
  * Returns pointer to the function of the library
  */

src/include/common/pg_socket.h

@@ -73,72 +73,72 @@ typedef struct PgSocketCall
 	int			(*fcntl) (int socket, int cmd, ... /* arg */ );
 } PgSocketCall;
 
-extern PgSocketCall *rcalls;
+extern PgSocketCall *pg_sock_vtab(void);
 
 /*
  * These macroses call socket function depending on isRsocket value
  */
 
 #define pg_fcntl(fd, flag, value, isRsocket) \
-	((isRsocket) ? rcalls->fcntl(fd, flag, value) : \
+	((isRsocket) ? pg_sock_vtab()->fcntl(fd, flag, value) : \
 		fcntl(fd, flag, value))
 
 #define pg_socket(domain, type, protocol, isRsocket) \
-	((isRsocket) ? rcalls->socket(domain, type, protocol) : \
+	((isRsocket) ? pg_sock_vtab()->socket(domain, type, protocol) : \
 		socket(domain, type, protocol))
 
 #define pg_bind(socket, addr, addrlen, isRsocket) \
-	((isRsocket) ? rcalls->bind(socket, addr, addrlen) : \
+	((isRsocket) ? pg_sock_vtab()->bind(socket, addr, addrlen) : \
 		bind(socket, addr, addrlen))
 
 #define pg_listen(socket, backlog, isRsocket) \
-	((isRsocket) ? rcalls->listen(socket, backlog) : \
+	((isRsocket) ? pg_sock_vtab()->listen(socket, backlog) : \
 		listen(socket, backlog))
 
 #define pg_accept(socket, addr, addrlen, isRsocket) \
-	((isRsocket) ? rcalls->accept(socket, addr, addrlen) : \
+	((isRsocket) ? pg_sock_vtab()->accept(socket, addr, addrlen) : \
 		accept(socket, addr, addrlen))
 
 #define pg_connect(socket, addr, addrlen, isRsocket) \
-	((isRsocket) ? rcalls->connect(socket, addr, addrlen) : \
+	((isRsocket) ? pg_sock_vtab()->connect(socket, addr, addrlen) : \
 		connect(socket, addr, addrlen))
 
 #define pg_closesocket(socket, isRsocket) \
-	((isRsocket) ? rcalls->close(socket) : \
+	((isRsocket) ? pg_sock_vtab()->close(socket) : \
 		closesocket(socket))
 
 #define pg_recv(socket, buf, len, flags, isRsocket) \
-	((isRsocket) ? rcalls->recv(socket, buf, len, flags) : \
+	((isRsocket) ? pg_sock_vtab()->recv(socket, buf, len, flags) : \
 		recv(socket, buf, len, flags))
 
 #define pg_send(socket, buf, len, flags, isRsocket) \
-	((isRsocket) ? rcalls->send(socket, buf, len, flags) : \
+	((isRsocket) ? pg_sock_vtab()->send(socket, buf, len, flags) : \
 		send(socket, buf, len, flags))
 
 #define pg_sendmsg(socket, msg, flags, isRsocket) \
-	((isRsocket) ? rcalls->sendmsg(socket, msg, flags) : \
+	((isRsocket) ? pg_sock_vtab()->sendmsg(socket, msg, flags) : \
 		sendmsg(socket, msg, flags))
 
 #ifdef HAVE_POLL
 #define pg_poll(fds, nfds, timeout, isRsocket) \
-	((isRsocket) ? rcalls->poll(fds, nfds, timeout) : \
+	((isRsocket) ? pg_sock_vtab()->poll(fds, nfds, timeout) : \
 		poll(fds, nfds, timeout))
 #endif
 
 #define pg_select(nfds, readfds, writefds, exceptfds, timeout, isRsocket) \
-	((isRsocket) ? rcalls->select(nfds, readfds, writefds, exceptfds, timeout) : \
+	((isRsocket) ? pg_sock_vtab()->select(nfds, readfds, writefds, exceptfds, timeout) : \
 		select(nfds, readfds, writefds, exceptfds, timeout))
 
 #define pg_getsockname(socket, addr, addrlen, isRsocket) \
-	((isRsocket) ? rcalls->getsockname(socket, addr, addrlen) : \
+	((isRsocket) ? pg_sock_vtab()->getsockname(socket, addr, addrlen) : \
 		getsockname(socket, addr, addrlen))
 
 #define pg_setsockopt(socket, level, optname, optval, optlen, isRsocket) \
-	((isRsocket) ? rcalls->setsockopt(socket, level, optname, optval, optlen) : \
+	((isRsocket) ? pg_sock_vtab()->setsockopt(socket, level, optname, optval, optlen) : \
 		setsockopt(socket, level, optname, optval, optlen))
 
 #define pg_getsockopt(socket, level, optname, optval, optlen, isRsocket) \
-	((isRsocket) ? rcalls->getsockopt(socket, level, optname, optval, optlen) : \
+	((isRsocket) ? pg_sock_vtab()->getsockopt(socket, level, optname, optval, optlen) : \
 		getsockopt(socket, level, optname, optval, optlen))
 
 /* port/pg_rsocket.c */