On Thu, Jan 10, 2019 at 01:10:16PM +0300, Sergei Kornilov wrote: > Without both ready_to_display and cleaning in RequestXLogStreaming > we do not show outdated PrimaryConnInfo in case walreceiver just > started, but does not connected yet? While waiting walrcv_connect > for example.
Good point. There is a gap between the moment the WAL receiver PID is set when the WAL receiver starts up and the moment where the different fields are reset to 0 which is not good as it could be possible that the user sees the information from the previous WAL receiver, so we should move the memset calls when the PID is set, reaching a state where the PID is alive but where there is no connection yet. The port number needs a similar treatment. Looking closer at the code, it seems to me that it would be a good thing if we update the shared memory state when the WAL receiver dies, so as not only the PID is set to 0, but all connection-related information gets the call. With all those comments I am finishing with the updated attached. Thoughts? -- Michael
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 9823b75767..4191b23621 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -11857,8 +11857,7 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
tli, curFileTLI);
}
curFileTLI = tli;
- RequestXLogStreaming(tli, ptr, PrimaryConnInfo,
- PrimarySlotName);
+ RequestXLogStreaming(tli, ptr);
receivedUpto = 0;
}
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index 2e90944ad5..7e3ff63a44 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -188,9 +188,7 @@ DisableWalRcvImmediateExit(void)
void
WalReceiverMain(void)
{
- char conninfo[MAXCONNINFO];
char *tmp_conninfo;
- char slotname[NAMEDATALEN];
XLogRecPtr startpoint;
TimeLineID startpointTLI;
TimeLineID primaryTLI;
@@ -248,10 +246,6 @@ WalReceiverMain(void)
walrcv->pid = MyProcPid;
walrcv->walRcvState = WALRCV_STREAMING;
- /* Fetch information required to start streaming */
- walrcv->ready_to_display = false;
- strlcpy(conninfo, (char *) walrcv->conninfo, MAXCONNINFO);
- strlcpy(slotname, (char *) walrcv->slotname, NAMEDATALEN);
startpoint = walrcv->receiveStart;
startpointTLI = walrcv->receiveStartTLI;
@@ -262,6 +256,16 @@ WalReceiverMain(void)
/* Report the latch to use to awaken this process */
walrcv->latch = &MyProc->procLatch;
+ /*
+ * Reset all connection information when the PID is set, which makes
+ * the information visible at SQL level, still we are not connected
+ * yet.
+ */
+ memset(walrcv->conninfo, 0, MAXCONNINFO);
+ memset(walrcv->slotname, 0, NAMEDATALEN);
+ memset(walrcv->sender_host, 0, NI_MAXHOST);
+ walrcv->sender_port = 0;
+
SpinLockRelease(&walrcv->mutex);
/* Arrange to clean up at walreceiver exit */
@@ -291,32 +295,36 @@ WalReceiverMain(void)
/* Unblock signals (they were blocked when the postmaster forked us) */
PG_SETMASK(&UnBlockSig);
+ if (PrimaryConnInfo == NULL || PrimaryConnInfo[0] == '\0')
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot connect to the primary server as \"primary_conninfo\" is not defined")));
+
/* Establish the connection to the primary for XLOG streaming */
EnableWalRcvImmediateExit();
- wrconn = walrcv_connect(conninfo, false, "walreceiver", &err);
+ wrconn = walrcv_connect(PrimaryConnInfo, false, "walreceiver", &err);
if (!wrconn)
ereport(ERROR,
(errmsg("could not connect to the primary server: %s", err)));
DisableWalRcvImmediateExit();
/*
- * Save user-visible connection string. This clobbers the original
- * conninfo, for security. Also save host and port of the sender server
- * this walreceiver is connected to.
+ * Save user-visible connection string. Also save host and port of the
+ * sender server this walreceiver is connected to.
*/
tmp_conninfo = walrcv_get_conninfo(wrconn);
walrcv_get_senderinfo(wrconn, &sender_host, &sender_port);
SpinLockAcquire(&walrcv->mutex);
- memset(walrcv->conninfo, 0, MAXCONNINFO);
if (tmp_conninfo)
strlcpy((char *) walrcv->conninfo, tmp_conninfo, MAXCONNINFO);
- memset(walrcv->sender_host, 0, NI_MAXHOST);
+ if (PrimarySlotName)
+ strlcpy((char *) walrcv->slotname, PrimarySlotName, NAMEDATALEN);
+
if (sender_host)
strlcpy((char *) walrcv->sender_host, sender_host, NI_MAXHOST);
walrcv->sender_port = sender_port;
- walrcv->ready_to_display = true;
SpinLockRelease(&walrcv->mutex);
if (tmp_conninfo)
@@ -387,7 +395,8 @@ WalReceiverMain(void)
*/
options.logical = false;
options.startpoint = startpoint;
- options.slotname = slotname[0] != '\0' ? slotname : NULL;
+ options.slotname = (PrimarySlotName && PrimarySlotName[0] != '\0') ?
+ PrimarySlotName : NULL;
options.proto.physical.startpointTLI = startpointTLI;
ThisTimeLineID = startpointTLI;
if (walrcv_startstreaming(wrconn, &options))
@@ -776,8 +785,17 @@ WalRcvDie(int code, Datum arg)
Assert(walrcv->pid == MyProcPid);
walrcv->walRcvState = WALRCV_STOPPED;
walrcv->pid = 0;
- walrcv->ready_to_display = false;
walrcv->latch = NULL;
+
+ /*
+ * Reset all connection information when the PID is reset, which makes
+ * the shared memory state consistent with the rest, even if this it not
+ * visible at SQL level.
+ */
+ memset(walrcv->conninfo, 0, MAXCONNINFO);
+ memset(walrcv->slotname, 0, NAMEDATALEN);
+ memset(walrcv->sender_host, 0, NI_MAXHOST);
+ walrcv->sender_port = 0;
SpinLockRelease(&walrcv->mutex);
/* Terminate the connection gracefully. */
@@ -1374,7 +1392,6 @@ pg_stat_get_wal_receiver(PG_FUNCTION_ARGS)
Datum *values;
bool *nulls;
int pid;
- bool ready_to_display;
WalRcvState state;
XLogRecPtr receive_start_lsn;
TimeLineID receive_start_tli;
@@ -1392,7 +1409,6 @@ pg_stat_get_wal_receiver(PG_FUNCTION_ARGS)
/* Take a lock to ensure value consistency */
SpinLockAcquire(&WalRcv->mutex);
pid = (int) WalRcv->pid;
- ready_to_display = WalRcv->ready_to_display;
state = WalRcv->walRcvState;
receive_start_lsn = WalRcv->receiveStart;
receive_start_tli = WalRcv->receiveStartTLI;
@@ -1412,7 +1428,7 @@ pg_stat_get_wal_receiver(PG_FUNCTION_ARGS)
* No WAL receiver (or not ready yet), just return a tuple with NULL
* values
*/
- if (pid == 0 || !ready_to_display)
+ if (pid == 0)
PG_RETURN_NULL();
/* determine result type */
diff --git a/src/backend/replication/walreceiverfuncs.c b/src/backend/replication/walreceiverfuncs.c
index 2d6cdfe0a2..68d37eb2c8 100644
--- a/src/backend/replication/walreceiverfuncs.c
+++ b/src/backend/replication/walreceiverfuncs.c
@@ -220,8 +220,7 @@ ShutdownWalRcv(void)
* of a replication slot to acquire.
*/
void
-RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr, const char *conninfo,
- const char *slotname)
+RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr)
{
WalRcvData *walrcv = WalRcv;
bool launch = false;
@@ -243,16 +242,6 @@ RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr, const char *conninfo,
Assert(walrcv->walRcvState == WALRCV_STOPPED ||
walrcv->walRcvState == WALRCV_WAITING);
- if (conninfo != NULL)
- strlcpy((char *) walrcv->conninfo, conninfo, MAXCONNINFO);
- else
- walrcv->conninfo[0] = '\0';
-
- if (slotname != NULL)
- strlcpy((char *) walrcv->slotname, slotname, NAMEDATALEN);
- else
- walrcv->slotname[0] = '\0';
-
if (walrcv->walRcvState == WALRCV_STOPPED)
{
launch = true;
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index e04d725ff5..cb62f40c07 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -104,8 +104,7 @@ typedef struct
TimestampTz latestWalEndTime;
/*
- * connection string; initially set to connect to the primary, and later
- * clobbered to hide security-sensitive fields.
+ * Connection string clobbered to hide security-sensitive fields.
*/
char conninfo[MAXCONNINFO];
@@ -122,9 +121,6 @@ typedef struct
*/
char slotname[NAMEDATALEN];
- /* set true once conninfo is ready to display (obfuscated pwds etc) */
- bool ready_to_display;
-
/*
* Latch used by startup process to wake up walreceiver after telling it
* where to start streaming (after setting receiveStart and
@@ -306,8 +302,7 @@ extern void WalRcvShmemInit(void);
extern void ShutdownWalRcv(void);
extern bool WalRcvStreaming(void);
extern bool WalRcvRunning(void);
-extern void RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr,
- const char *conninfo, const char *slotname);
+extern void RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr);
extern XLogRecPtr GetWalRcvWriteRecPtr(XLogRecPtr *latestChunkStart, TimeLineID *receiveTLI);
extern int GetReplicationApplyDelay(void);
extern int GetReplicationTransferLatency(void);
signature.asc
Description: PGP signature
