Skip to content

Commit 2ec9317

Browse files
dmpgprodmpgpro
committed
Merge branch 'PGPROEE9_6' into PGPROEE9_6_RUSDOC
2 parents 4a1d354 + a0d4a8a commit 2ec9317

File tree

199 files changed

+7817
-2612
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

199 files changed

+7817
-2612
lines changed

‎.ci/run_install_check

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,11 +25,13 @@ LD_LIBRARY_PATH=$(pwd)/tmp_install/lib
2525
DYLD_LIBRARY_PATH=$(pwd)/tmp_install/lib
2626
PATH=$(pwd)/tmp_install/bin:${PATH}
2727
PGDATA=$(pwd)/tmp_base
28-
export DYLD_LIBRARY_PATH LD_LIBRARY_PATH PATH PG_DATA
28+
export DYLD_LIBRARY_PATH LD_LIBRARY_PATH PATH PGDATA
2929

3030
# create installation
3131
PGPORT=`./.ci/find_free_port 5432`
32-
export PGPORT
32+
LANG=C
33+
LC_ALL=C
34+
export PGPORT LANG LC_ALL
3335
./.ci/make_test_base $PGDATA
3436
#run checks
3537
set +e

‎configure

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2770,7 +2770,7 @@ else
27702770
fi
27712771

27722772

2773-
PGPRO_VERSION="$PACKAGE_VERSION.2"
2773+
PGPRO_VERSION="$PACKAGE_VERSION.3"
27742774
PGPRO_PACKAGE_NAME="PostgresPro"
27752775
PGPRO_EDITION="enterprise"
27762776

@@ -14869,7 +14869,7 @@ fi
1486914869
LIBS_including_readline="$LIBS"
1487014870
LIBS=`echo "$LIBS" | sed -e 's/-ledit//g' -e 's/-lreadline//g'`
1487114871

14872-
for ac_func in cbrt dlopen fdatasync getifaddrs getpeerucred getrlimit mbstowcs_l memmove poll pstat pthread_is_threaded_np readlink setproctitle setsid shm_open symlink sync_file_range towlower utime utimes wcstombs wcstombs_l
14872+
for ac_func in cbrt dlopen fdatasync getifaddrs getpeerucred getrlimit mbstowcs_l memmove poll posix_fallocate pstat pthread_is_threaded_np readlink setproctitle setsid shm_open symlink sync_file_range towlower utime utimes wcstombs wcstombs_l
1487314873
do :
1487414874
as_ac_var=`$as_echo "ac_cv_func_$ac_func" | $as_tr_sh`
1487514875
ac_fn_c_check_func "$LINENO" "$ac_func" "$as_ac_var"

‎configure.in

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ AC_DEFINE_UNQUOTED(PG_MAJORVERSION, "$PG_MAJORVERSION", [PostgreSQL major versio
3838
PGAC_ARG_REQ(with, extra-version, [STRING], [append STRING to version],
3939
[PG_VERSION="$PACKAGE_VERSION$withval"],
4040
[PG_VERSION="$PACKAGE_VERSION"])
41-
PGPRO_VERSION="$PACKAGE_VERSION.2"
41+
PGPRO_VERSION="$PACKAGE_VERSION.3"
4242
PGPRO_PACKAGE_NAME="PostgresPro"
4343
PGPRO_EDITION="enterprise"
4444
AC_SUBST(PGPRO_PACKAGE_NAME)
@@ -1583,7 +1583,7 @@ PGAC_FUNC_WCSTOMBS_L
15831583
LIBS_including_readline="$LIBS"
15841584
LIBS=`echo "$LIBS" | sed -e 's/-ledit//g' -e 's/-lreadline//g'`
15851585

1586-
AC_CHECK_FUNCS([cbrt dlopen fdatasync getifaddrs getpeerucred getrlimit mbstowcs_l memmove poll pstat pthread_is_threaded_np readlink setproctitle setsid shm_open symlink sync_file_range towlower utime utimes wcstombs wcstombs_l])
1586+
AC_CHECK_FUNCS([cbrt dlopen fdatasync getifaddrs getpeerucred getrlimit mbstowcs_l memmove poll posix_fallocate pstat pthread_is_threaded_np readlink setproctitle setsid shm_open symlink sync_file_range towlower utime utimes wcstombs wcstombs_l])
15871587

15881588
AC_REPLACE_FUNCS(fseeko)
15891589
case $host_os in

‎contrib/Makefile

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@ SUBDIRS = \
4747
pg_visibility \
4848
pg_wait_sampling \
4949
postgres_fdw \
50+
referee \
5051
rum \
5152
seg \
5253
spi \

‎contrib/mmts/Cluster.pm

Lines changed: 17 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,7 @@ sub new
6666
my $node = new PostgresNode("node$i", $host, $pgport);
6767
$node->{id} = $i;
6868
$node->{arbiter_port} = $arbiter_port;
69+
$node->{mmconnstr} = "${ \$node->connstr('postgres') } arbiter_port=${ \$node->{arbiter_port} }";
6970
push(@$nodes, $node);
7071
}
7172

@@ -89,47 +90,54 @@ sub init
8990
}
9091
}
9192

93+
sub all_connstrs
94+
{
95+
my ($self) = @_;
96+
my $nodes = $self->{nodes};
97+
return join(', ', map { "${ \$_->connstr('postgres') } arbiter_port=${ \$_->{arbiter_port} }" } @$nodes);
98+
}
99+
100+
92101
sub configure
93102
{
94103
my ($self) = @_;
95104
my $nodes = $self->{nodes};
96-
my $nnodes = scalar @{ $nodes };
97105

98-
my $connstr = join(', ', map { "${ \$_->connstr('postgres') } arbiter_port=${ \$_->{arbiter_port} }" } @$nodes);
106+
my $connstr = $self->all_connstrs();
99107

100108
foreach my $node (@$nodes)
101109
{
102110
my $id = $node->{id};
103111
my $host = $node->host;
104112
my $pgport = $node->port;
105113
my $arbiter_port = $node->{arbiter_port};
114+
my $unix_sock_dir = $ENV{PGHOST};
106115

107116
$node->append_conf("postgresql.conf", qq(
108117
log_statement = none
109118
listen_addresses = '$host'
110-
unix_socket_directories = ''
119+
unix_socket_directories = '$unix_sock_dir'
111120
port = $pgport
112121
max_prepared_transactions = 10
113122
max_connections = 10
114123
max_worker_processes = 100
115124
wal_level = logical
116-
max_wal_senders = 5
125+
max_wal_senders = 6
117126
wal_sender_timeout = 0
118127
default_transaction_isolation = 'repeatable read'
119-
max_replication_slots = 5
128+
max_replication_slots = 6
120129
shared_preload_libraries = 'multimaster'
121130
shared_buffers = 16MB
122131
123132
multimaster.arbiter_port = $arbiter_port
124133
multimaster.workers = 1
125134
multimaster.node_id = $id
126135
multimaster.conn_strings = '$connstr'
127-
multimaster.heartbeat_recv_timeout = 2050
136+
multimaster.heartbeat_recv_timeout = 1050
128137
multimaster.heartbeat_send_timeout = 250
129-
multimaster.max_nodes = $nnodes
130-
multimaster.ignore_tables_without_pk = true
138+
multimaster.max_nodes = 6
139+
multimaster.ignore_tables_without_pk = false
131140
multimaster.queue_size = 4194304
132-
multimaster.min_2pc_timeout = 150000
133141
log_line_prefix = '%t: '
134142
));
135143

‎contrib/mmts/Dockerfile

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,9 @@ COPY ./ /pg/mmts/
66
RUN export USE_PGXS=1 && \
77
cd /pg/mmts && make clean && make install
88

9+
RUN export USE_PGXS=1 && \
10+
cd /pg/src/contrib/referee && make clean && make install
11+
912
# pg_regress client assumes such dir exists on server
1013
RUN cp /pg/src/src/test/regress/*.so /pg/install/lib/postgresql/
1114
USER postgres

‎contrib/mmts/Makefile

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11

22
EXTENSION = multimaster
33
DATA = multimaster--1.0.sql
4-
OBJS = multimaster.o arbiter.o bytebuf.o bgwpool.o pglogical_output.o pglogical_proto.o pglogical_receiver.o pglogical_apply.o pglogical_hooks.o pglogical_config.o pglogical_relid_map.o ddd.o bkb.o spill.o referee.o
4+
OBJS = multimaster.o arbiter.o bytebuf.o bgwpool.o pglogical_output.o pglogical_proto.o pglogical_receiver.o pglogical_apply.o pglogical_hooks.o pglogical_config.o pglogical_relid_map.o ddd.o bkb.o spill.o referee.o state.o
55
MODULE_big = multimaster
66

77
PG_CPPFLAGS = -I$(libpq_srcdir)

‎contrib/mmts/arbiter.c

Lines changed: 56 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,7 @@
7676

7777

7878
#include "multimaster.h"
79+
#include "state.h"
7980

8081
#define MAX_ROUTES 16
8182
#define INIT_BUFFER_SIZE 1024
@@ -189,7 +190,6 @@ static void MtmDisconnect(int node)
189190
MtmUnregisterSocket(sockets[node]);
190191
pg_closesocket(sockets[node], MtmUseRDMA);
191192
sockets[node] = -1;
192-
MtmOnNodeDisconnect(node+1);
193193
}
194194

195195
static int MtmWaitSocket(int sd, bool forWrite, timestamp_t timeoutMsec)
@@ -316,25 +316,22 @@ static void MtmCheckResponse(MtmArbiterMessage* resp)
316316
} else {
317317
BIT_CLEAR(Mtm->currentLockNodeMask, resp->node-1);
318318
}
319-
if (
320-
( BIT_CHECK(resp->disabledNodeMask, MtmNodeId-1) || Mtm->status == MTM_IN_MINORITY )
321-
&& !BIT_CHECK(Mtm->disabledNodeMask, resp->node-1)
322-
&& Mtm->status != MTM_RECOVERY
323-
&& Mtm->status != MTM_RECOVERED
324-
&& Mtm->nodes[MtmNodeId-1].lastStatusChangeTime + MSEC_TO_USEC(MtmNodeDisableDelay) < MtmGetSystemTime())
325-
{
326-
MTM_ELOG(WARNING, "Node %d thinks that I'm dead, while I'm %s (message %s)", resp->node, MtmNodeStatusMnem[Mtm->status], MtmMessageKindMnem[resp->code]);
327-
BIT_SET(Mtm->disabledNodeMask, MtmNodeId-1);
328-
Mtm->nConfigChanges += 1;
329-
MtmSwitchClusterMode(MTM_RECOVERY);
330-
} else if (BIT_CHECK(Mtm->disabledNodeMask, resp->node-1) && sockets[resp->node-1] < 0) {
331-
/* We receive heartbeat from disabled node.
319+
320+
// if (BIT_CHECK(resp->disabledNodeMask, MtmNodeId-1))
321+
// {
322+
// MtmStateProcessEvent(MTM_REMOTE_DISABLE);
323+
// }
324+
325+
if (BIT_CHECK(Mtm->disabledNodeMask, resp->node-1) &&
326+
sockets[resp->node-1] < 0)
327+
{
328+
/* We've received heartbeat from disabled node.
332329
* Looks like it is restarted.
333330
* Try to reconnect to it.
334331
*/
335332
MTM_ELOG(WARNING, "Receive heartbeat from disabled node %d", resp->node);
336333
BIT_SET(Mtm->reconnectMask, resp->node-1);
337-
}
334+
}
338335
}
339336

340337
static void MtmScheduleHeartbeat()
@@ -362,11 +359,17 @@ static void MtmSendHeartbeat()
362359
for (i = 0; i < Mtm->nAllNodes; i++)
363360
{
364361
if (i+1 != MtmNodeId) {
365-
if (!BIT_CHECK(busy_mask, i)
366-
&& (Mtm->status != MTM_ONLINE
367-
|| sockets[i] >= 0
368-
|| !BIT_CHECK(Mtm->disabledNodeMask, i)
369-
|| BIT_CHECK(Mtm->reconnectMask, i)))
362+
if (!BIT_CHECK(busy_mask, i))
363+
/*
364+
* Old behaviour here can cause subtle bugs, for example
365+
* it can happened that none of mentioned conditiotions is
366+
* true when disabled node connects to a major node which
367+
* is online. So just send it allways. --sk
368+
*/
369+
// && (Mtm->status != MTM_ONLINE
370+
// || sockets[i] >= 0
371+
// || !BIT_CHECK(Mtm->disabledNodeMask, i)
372+
// || BIT_CHECK(Mtm->reconnectMask, i)))
370373
{
371374
if (!MtmSendToNode(i, &msg, sizeof(msg))) {
372375
MTM_ELOG(LOG, "Arbiter failed to send heartbeat to node %d", i+1);
@@ -543,17 +546,9 @@ static void MtmOpenConnections()
543546
for (i = 0; i < nNodes; i++) {
544547
if (i+1 != MtmNodeId && i < Mtm->nAllNodes) {
545548
sockets[i] = MtmConnectSocket(i, Mtm->nodes[i].con.arbiterPort);
546-
if (sockets[i] < 0) {
547-
MtmOnNodeDisconnect(i+1);
548-
}
549549
}
550550
}
551-
if (Mtm->nLiveNodes < Mtm->nAllNodes/2+1) { /* no quorum */
552-
MTM_ELOG(WARNING, "Node is out of quorum: only %d nodes of %d are accessible", Mtm->nLiveNodes, Mtm->nAllNodes);
553-
MtmSwitchClusterMode(MTM_IN_MINORITY);
554-
} else if (Mtm->status == MTM_INITIALIZATION) {
555-
MtmSwitchClusterMode(MTM_CONNECTED);
556-
}
551+
MtmStateProcessEvent(MTM_ARBITER_RECEIVER_START);
557552
}
558553

559554

@@ -586,7 +581,6 @@ static bool MtmSendToNode(int node, void const* buf, int size)
586581
}
587582
sockets[node] = MtmConnectSocket(node, Mtm->nodes[node].con.arbiterPort);
588583
if (sockets[node] < 0) {
589-
MtmOnNodeDisconnect(node+1);
590584
result = false;
591585
break;
592586
}
@@ -716,16 +710,18 @@ static void MtmSender(Datum arg)
716710
{
717711
int nNodes = MtmMaxNodes;
718712
int i;
713+
MtmBuffer* txBuffer;
719714

720715
MtmBackgroundWorker = true;
721716

722-
MtmBuffer* txBuffer = (MtmBuffer*)palloc0(sizeof(MtmBuffer)*nNodes);
717+
txBuffer = (MtmBuffer*)palloc0(sizeof(MtmBuffer)*nNodes);
723718
MTM_ELOG(LOG, "Start arbiter sender %d", MyProcPid);
724719
InitializeTimeouts();
725720

726721
pqsignal(SIGINT, SetStop);
727722
pqsignal(SIGQUIT, SetStop);
728723
pqsignal(SIGTERM, SetStop);
724+
pqsignal(SIGHUP, PostgresSigHupHandler);
729725

730726
/* We're now ready to receive signals */
731727
BackgroundWorkerUnblockSignals();
@@ -744,6 +740,12 @@ static void MtmSender(Datum arg)
744740
PGSemaphoreLock(&Mtm->sendSemaphore);
745741
CHECK_FOR_INTERRUPTS();
746742

743+
if (ConfigReloadPending)
744+
{
745+
ConfigReloadPending = false;
746+
ProcessConfigFile(PGC_SIGHUP);
747+
}
748+
747749
MtmCheckHeartbeat();
748750
/*
749751
* Use shared lock to improve locality,
@@ -805,6 +807,7 @@ static void MtmMonitor(Datum arg)
805807
pqsignal(SIGINT, SetStop);
806808
pqsignal(SIGQUIT, SetStop);
807809
pqsignal(SIGTERM, SetStop);
810+
pqsignal(SIGHUP, PostgresSigHupHandler);
808811

809812
MtmBackgroundWorker = true;
810813

@@ -819,6 +822,13 @@ static void MtmMonitor(Datum arg)
819822
if (rc & WL_POSTMASTER_DEATH) {
820823
break;
821824
}
825+
826+
if (ConfigReloadPending)
827+
{
828+
ConfigReloadPending = false;
829+
ProcessConfigFile(PGC_SIGHUP);
830+
}
831+
822832
MtmRefreshClusterStatus();
823833
}
824834
}
@@ -844,6 +854,7 @@ static void MtmReceiver(Datum arg)
844854
pqsignal(SIGINT, SetStop);
845855
pqsignal(SIGQUIT, SetStop);
846856
pqsignal(SIGTERM, SetStop);
857+
pqsignal(SIGHUP, PostgresSigHupHandler);
847858

848859
MtmBackgroundWorker = true;
849860

@@ -879,7 +890,14 @@ static void MtmReceiver(Datum arg)
879890
for (j = 0; j < n; j++) {
880891
if (events[j].events & EPOLLIN)
881892
#else
882-
fd_set events;
893+
fd_set events;
894+
895+
if (ConfigReloadPending)
896+
{
897+
ConfigReloadPending = false;
898+
ProcessConfigFile(PGC_SIGHUP);
899+
}
900+
883901
do {
884902
struct timeval tv;
885903
events = inset;
@@ -969,6 +987,7 @@ static void MtmReceiver(Datum arg)
969987
msg->gid, MtmTxnStatusMnem[msg->status], node);
970988

971989
replorigin_session_origin = DoNotReplicateId;
990+
TXFINISH("%s ABORT, MSG_POLL_STATUS", msg->gid);
972991
MtmFinishPreparedTransaction(ts, false);
973992
replorigin_session_origin = InvalidRepOriginId;
974993
}
@@ -982,6 +1001,7 @@ static void MtmReceiver(Datum arg)
9821001
MTM_ELOG(LOG, "Commit transaction %s because it is prepared at all live nodes", msg->gid);
9831002

9841003
replorigin_session_origin = DoNotReplicateId;
1004+
TXFINISH("%s COMMIT, MSG_POLL_STATUS", msg->gid);
9851005
MtmFinishPreparedTransaction(ts, true);
9861006
replorigin_session_origin = InvalidRepOriginId;
9871007
} else {
@@ -1006,7 +1026,7 @@ static void MtmReceiver(Datum arg)
10061026
default:
10071027
break;
10081028
}
1009-
if (BIT_CHECK(msg->disabledNodeMask, node-1)) {
1029+
if (BIT_CHECK(msg->disabledNodeMask, node-1) || BIT_CHECK(Mtm->disabledNodeMask, node-1)) {
10101030
MTM_ELOG(WARNING, "Ignore message from dead node %d\n", node);
10111031
continue;
10121032
}
@@ -1057,17 +1077,10 @@ static void MtmReceiver(Datum arg)
10571077
if (ts->isTwoPhase) {
10581078
MtmWakeUpBackend(ts);
10591079
} else if (MtmUseDtm) {
1060-
ts->votedMask = 0;
10611080
MTM_TXTRACE(ts, "MtmTransReceiver send MSG_PRECOMMIT");
10621081
Assert(replorigin_session_origin == InvalidRepOriginId);
1063-
MTM_LOG2("SetPreparedTransactionState for %s", ts->gid);
1064-
MtmUnlock();
1065-
MtmResetTransaction();
1066-
StartTransactionCommand();
1067-
SetPreparedTransactionState(ts->gid, MULTIMASTER_PRECOMMITTED);
1068-
CommitTransactionCommand();
1069-
Assert(!MtmTransIsActive());
1070-
MtmLock(LW_EXCLUSIVE);
1082+
ts->isPrepared = false;
1083+
SetLatch(&ProcGlobal->allProcs[ts->procno].procLatch);
10711084
} else {
10721085
ts->status = TRANSACTION_STATUS_UNKNOWN;
10731086
MtmWakeUpBackend(ts);
@@ -1084,7 +1097,7 @@ static void MtmReceiver(Datum arg)
10841097
if (ts->status != TRANSACTION_STATUS_ABORTED) {
10851098
MTM_LOG1("Arbiter receive abort message for transaction %s (%llu) from node %d", ts->gid, (long64)ts->xid, node);
10861099
Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
1087-
ts->aborted_by_node = node;
1100+
ts->abortedByNode = node;
10881101
MtmAbortTransaction(ts);
10891102
}
10901103
if ((ts->participantsMask & ~Mtm->disabledNodeMask & ~ts->votedMask) == 0) {
@@ -1161,4 +1174,3 @@ static void MtmReceiver(Datum arg)
11611174
}
11621175
proc_exit(1); /* force restart of this bgwroker */
11631176
}
1164-

0 commit comments

Comments
 (0)