Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

Commit c4f99d2

Browse filesBrowse files
MasaoFujiiMasaoFujii
committed
Add --synchronous option to pg_receivexlog, for more reliable WAL writing.
Previously pg_receivexlog flushed WAL data only when WAL file was switched. Then 3dad73e added -F option to pg_receivexlog so that users could control how frequently sync commands were issued to WAL files. It also allowed users to make pg_receivexlog flush WAL data immediately after writing by specifying 0 in -F option. However feedback messages were not sent back immediately even after a flush location was updated. So even if WAL data was flushed in real time, the server could not see that for a while. This commit removes -F option from and adds --synchronous to pg_receivexlog. If --synchronous is specified, like the standby's wal receiver, pg_receivexlog flushes WAL data as soon as there is WAL data which has not been flushed yet. Then it sends back the feedback message identifying the latest flush location to the server. This option is useful to make pg_receivexlog behave as sync standby by using replication slot, for example. Original patch by Furuya Osamu, heavily rewritten by me. Reviewed by Heikki Linnakangas, Alvaro Herrera and Sawada Masahiko.
1 parent bc24148 commit c4f99d2
Copy full SHA for c4f99d2

File tree

Expand file treeCollapse file tree

5 files changed

+54
-72
lines changed
Filter options
Expand file treeCollapse file tree

5 files changed

+54
-72
lines changed

‎doc/src/sgml/ref/pg_receivexlog.sgml

Copy file name to clipboardExpand all lines: doc/src/sgml/ref/pg_receivexlog.sgml
+22-20Lines changed: 22 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,13 @@ PostgreSQL documentation
4848
<application>pg_receivexlog</application>.
4949
</para>
5050

51+
<para>
52+
Unlike the standby's WAL receiver, <application>pg_receivexlog</>
53+
flushes WAL data only when WAL file is closed, by default.
54+
<literal>--synchronous</> option must be specified to flush WAL data
55+
in real time and ensure it's safely flushed to disk.
56+
</para>
57+
5158
<para>
5259
The transaction log is streamed over a regular
5360
<productname>PostgreSQL</productname> connection, and uses the replication
@@ -85,21 +92,6 @@ PostgreSQL documentation
8592
</listitem>
8693
</varlistentry>
8794

88-
<varlistentry>
89-
<term><option>-F <replaceable class="parameter">interval</replaceable></option></term>
90-
<term><option>--fsync-interval=<replaceable class="parameter">interval</replaceable></option></term>
91-
<listitem>
92-
<para>
93-
Specifies the maximum time to issue sync commands to ensure the
94-
received WAL file is safely flushed to disk, in seconds. The default
95-
value is zero, which disables issuing fsyncs except when WAL file is
96-
closed. If <literal>-1</literal> is specified, WAL file is flushed as
97-
soon as possible, that is, as soon as there are WAL data which has
98-
not been flushed yet.
99-
</para>
100-
</listitem>
101-
</varlistentry>
102-
10395
<varlistentry>
10496
<term><option>-n</option></term>
10597
<term><option>--no-loop</option></term>
@@ -135,15 +127,25 @@ PostgreSQL documentation
135127
When this option is used, <application>pg_receivexlog</> will report
136128
a flush position to the server, indicating when each segment has been
137129
synchronized to disk so that the server can remove that segment if it
138-
is not otherwise needed. When using this parameter, it is important
139-
to make sure that <application>pg_receivexlog</> cannot become the
140-
synchronous standby through an incautious setting of
141-
<xref linkend="guc-synchronous-standby-names">; it does not flush
142-
data frequently enough for this to work correctly.
130+
is not otherwise needed. <literal>--synchronous</literal> option must
131+
be specified when making <application>pg_receivexlog</> run as
132+
synchronous standby by using replication slot. Otherwise WAL data
133+
cannot be flushed frequently enough for this to work correctly.
143134
</para>
144135
</listitem>
145136
</varlistentry>
146137

138+
<varlistentry>
139+
<term><option>--synchronous</option></term>
140+
<listitem>
141+
<para>
142+
Issue sync commands as soon as there is WAL data which has not been
143+
flushed yet. Also status packets are sent back to the server just after
144+
WAL data is flushed whatever <literal>--status-interval</> is set to.
145+
</para>
146+
</listitem>
147+
</varlistentry>
148+
147149
<varlistentry>
148150
<term><option>-v</option></term>
149151
<term><option>--verbose</option></term>

‎src/bin/pg_basebackup/pg_basebackup.c

Copy file name to clipboardExpand all lines: src/bin/pg_basebackup/pg_basebackup.c
+1-1Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -370,7 +370,7 @@ LogStreamerMain(logstreamer_param *param)
370370
if (!ReceiveXlogStream(param->bgconn, param->startptr, param->timeline,
371371
param->sysidentifier, param->xlogdir,
372372
reached_end_position, standby_message_timeout,
373-
NULL, 0))
373+
NULL, false))
374374

375375
/*
376376
* Any errors will already have been reported in the function process,

‎src/bin/pg_basebackup/pg_receivexlog.c

Copy file name to clipboardExpand all lines: src/bin/pg_basebackup/pg_receivexlog.c
+8-15Lines changed: 8 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -36,10 +36,10 @@ static char *basedir = NULL;
3636
static int verbose = 0;
3737
static int noloop = 0;
3838
static int standby_message_timeout = 10 * 1000; /* 10 sec = default */
39-
static int fsync_interval = 0; /* 0 = default */
4039
static volatile bool time_to_abort = false;
4140
static bool do_create_slot = false;
4241
static bool do_drop_slot = false;
42+
static bool synchronous = false;
4343

4444

4545
static void usage(void);
@@ -66,12 +66,11 @@ usage(void)
6666
printf(_(" %s [OPTION]...\n"), progname);
6767
printf(_("\nOptions:\n"));
6868
printf(_(" -D, --directory=DIR receive transaction log files into this directory\n"));
69-
printf(_(" -F --fsync-interval=SECS\n"
70-
" time between fsyncs to transaction log files (default: %d)\n"), (fsync_interval / 1000));
7169
printf(_(" -n, --no-loop do not loop on connection lost\n"));
7270
printf(_(" -s, --status-interval=SECS\n"
7371
" time between status packets sent to server (default: %d)\n"), (standby_message_timeout / 1000));
7472
printf(_(" -S, --slot=SLOTNAME replication slot to use\n"));
73+
printf(_(" --synchronous flush transaction log immediately after writing\n"));
7574
printf(_(" -v, --verbose output verbose messages\n"));
7675
printf(_(" -V, --version output version information, then exit\n"));
7776
printf(_(" -?, --help show this help, then exit\n"));
@@ -343,7 +342,7 @@ StreamLog(void)
343342

344343
ReceiveXlogStream(conn, startpos, starttli, NULL, basedir,
345344
stop_streaming, standby_message_timeout, ".partial",
346-
fsync_interval);
345+
synchronous);
347346

348347
PQfinish(conn);
349348
conn = NULL;
@@ -374,7 +373,6 @@ main(int argc, char **argv)
374373
{"port", required_argument, NULL, 'p'},
375374
{"username", required_argument, NULL, 'U'},
376375
{"no-loop", no_argument, NULL, 'n'},
377-
{"fsync-interval", required_argument, NULL, 'F'},
378376
{"no-password", no_argument, NULL, 'w'},
379377
{"password", no_argument, NULL, 'W'},
380378
{"status-interval", required_argument, NULL, 's'},
@@ -383,6 +381,7 @@ main(int argc, char **argv)
383381
/* action */
384382
{"create-slot", no_argument, NULL, 1},
385383
{"drop-slot", no_argument, NULL, 2},
384+
{"synchronous", no_argument, NULL, 3},
386385
{NULL, 0, NULL, 0}
387386
};
388387

@@ -408,7 +407,7 @@ main(int argc, char **argv)
408407
}
409408
}
410409

411-
while ((c = getopt_long(argc, argv, "D:d:h:p:U:s:S:nF:wWv",
410+
while ((c = getopt_long(argc, argv, "D:d:h:p:U:s:S:nwWv",
412411
long_options, &option_index)) != -1)
413412
{
414413
switch (c)
@@ -455,15 +454,6 @@ main(int argc, char **argv)
455454
case 'n':
456455
noloop = 1;
457456
break;
458-
case 'F':
459-
fsync_interval = atoi(optarg) * 1000;
460-
if (fsync_interval < -1000)
461-
{
462-
fprintf(stderr, _("%s: invalid fsync interval \"%s\"\n"),
463-
progname, optarg);
464-
exit(1);
465-
}
466-
break;
467457
case 'v':
468458
verbose++;
469459
break;
@@ -474,6 +464,9 @@ main(int argc, char **argv)
474464
case 2:
475465
do_drop_slot = true;
476466
break;
467+
case 3:
468+
synchronous = true;
469+
break;
477470
default:
478471

479472
/*

‎src/bin/pg_basebackup/receivelog.c

Copy file name to clipboardExpand all lines: src/bin/pg_basebackup/receivelog.c
+22-35Lines changed: 22 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -31,14 +31,13 @@ static char current_walfile_name[MAXPGPATH] = "";
3131
static bool reportFlushPosition = false;
3232
static XLogRecPtr lastFlushPosition = InvalidXLogRecPtr;
3333

34-
static int64 last_fsync = -1; /* timestamp of last WAL file flush */
3534
static bool still_sending = true; /* feedback still needs to be sent? */
3635

3736
static PGresult *HandleCopyStream(PGconn *conn, XLogRecPtr startpos,
3837
uint32 timeline, char *basedir,
3938
stream_stop_callback stream_stop, int standby_message_timeout,
4039
char *partial_suffix, XLogRecPtr *stoppos,
41-
int fsync_interval);
40+
bool synchronous);
4241
static int CopyStreamPoll(PGconn *conn, long timeout_ms);
4342
static int CopyStreamReceive(PGconn *conn, long timeout, char **buffer);
4443
static bool ProcessKeepaliveMsg(PGconn *conn, char *copybuf, int len,
@@ -55,8 +54,7 @@ static bool CheckCopyStreamStop(PGconn *conn, XLogRecPtr blockpos,
5554
stream_stop_callback stream_stop,
5655
char *partial_suffix, XLogRecPtr *stoppos);
5756
static long CalculateCopyStreamSleeptime(int64 now, int standby_message_timeout,
58-
int64 last_status, int fsync_interval,
59-
XLogRecPtr blockpos);
57+
int64 last_status);
6058

6159
static bool ReadEndOfStreamingResult(PGresult *res, XLogRecPtr *startpos,
6260
uint32 *timeline);
@@ -209,7 +207,6 @@ close_walfile(char *basedir, char *partial_suffix, XLogRecPtr pos)
209207
progname, current_walfile_name, partial_suffix);
210208

211209
lastFlushPosition = pos;
212-
last_fsync = feGetCurrentTimestamp();
213210
return true;
214211
}
215212

@@ -440,8 +437,8 @@ CheckServerVersionForStreaming(PGconn *conn)
440437
* allows you to tell the difference between partial and completed files,
441438
* so that you can continue later where you left.
442439
*
443-
* fsync_interval controls how often we flush to the received WAL file,
444-
* in milliseconds.
440+
* If 'synchronous' is true, the received WAL is flushed as soon as written,
441+
* otherwise only when the WAL file is closed.
445442
*
446443
* Note: The log position *must* be at a log segment start!
447444
*/
@@ -450,7 +447,7 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
450447
char *sysidentifier, char *basedir,
451448
stream_stop_callback stream_stop,
452449
int standby_message_timeout, char *partial_suffix,
453-
int fsync_interval)
450+
bool synchronous)
454451
{
455452
char query[128];
456453
char slotcmd[128];
@@ -595,7 +592,7 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
595592
/* Stream the WAL */
596593
res = HandleCopyStream(conn, startpos, timeline, basedir, stream_stop,
597594
standby_message_timeout, partial_suffix,
598-
&stoppos, fsync_interval);
595+
&stoppos, synchronous);
599596
if (res == NULL)
600597
goto error;
601598

@@ -760,7 +757,7 @@ static PGresult *
760757
HandleCopyStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
761758
char *basedir, stream_stop_callback stream_stop,
762759
int standby_message_timeout, char *partial_suffix,
763-
XLogRecPtr *stoppos, int fsync_interval)
760+
XLogRecPtr *stoppos, bool synchronous)
764761
{
765762
char *copybuf = NULL;
766763
int64 last_status = -1;
@@ -784,24 +781,26 @@ HandleCopyStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
784781
now = feGetCurrentTimestamp();
785782

786783
/*
787-
* If fsync_interval has elapsed since last WAL flush and we've written
788-
* some WAL data, flush them to disk.
784+
* If synchronous option is true, issue sync command as soon as
785+
* there are WAL data which has not been flushed yet.
789786
*/
790-
if (lastFlushPosition < blockpos &&
791-
walfile != -1 &&
792-
((fsync_interval > 0 &&
793-
feTimestampDifferenceExceeds(last_fsync, now, fsync_interval)) ||
794-
fsync_interval < 0))
787+
if (synchronous && lastFlushPosition < blockpos && walfile != -1)
795788
{
796789
if (fsync(walfile) != 0)
797790
{
798791
fprintf(stderr, _("%s: could not fsync file \"%s\": %s\n"),
799792
progname, current_walfile_name, strerror(errno));
800793
goto error;
801794
}
802-
803795
lastFlushPosition = blockpos;
804-
last_fsync = now;
796+
797+
/*
798+
* Send feedback so that the server sees the latest WAL locations
799+
* immediately.
800+
*/
801+
if (!sendFeedback(conn, blockpos, now, false))
802+
goto error;
803+
last_status = now;
805804
}
806805

807806
/*
@@ -821,7 +820,7 @@ HandleCopyStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
821820
* Calculate how long send/receive loops should sleep
822821
*/
823822
sleeptime = CalculateCopyStreamSleeptime(now, standby_message_timeout,
824-
last_status, fsync_interval, blockpos);
823+
last_status);
825824

826825
r = CopyStreamReceive(conn, sleeptime, &copybuf);
827826
while (r != 0)
@@ -1244,34 +1243,22 @@ CheckCopyStreamStop(PGconn *conn, XLogRecPtr blockpos, uint32 timeline,
12441243
*/
12451244
static long
12461245
CalculateCopyStreamSleeptime(int64 now, int standby_message_timeout,
1247-
int64 last_status, int fsync_interval, XLogRecPtr blockpos)
1246+
int64 last_status)
12481247
{
1249-
int64 targettime = 0;
12501248
int64 status_targettime = 0;
1251-
int64 fsync_targettime = 0;
12521249
long sleeptime;
12531250

12541251
if (standby_message_timeout && still_sending)
12551252
status_targettime = last_status +
12561253
(standby_message_timeout - 1) * ((int64) 1000);
12571254

1258-
if (fsync_interval > 0 && lastFlushPosition < blockpos)
1259-
fsync_targettime = last_fsync +
1260-
(fsync_interval - 1) * ((int64) 1000);
1261-
1262-
if ((status_targettime < fsync_targettime && status_targettime > 0) ||
1263-
fsync_targettime == 0)
1264-
targettime = status_targettime;
1265-
else
1266-
targettime = fsync_targettime;
1267-
1268-
if (targettime > 0)
1255+
if (status_targettime > 0)
12691256
{
12701257
long secs;
12711258
int usecs;
12721259

12731260
feTimestampDifference(now,
1274-
targettime,
1261+
status_targettime,
12751262
&secs,
12761263
&usecs);
12771264
/* Always sleep at least 1 sec */

‎src/bin/pg_basebackup/receivelog.h

Copy file name to clipboardExpand all lines: src/bin/pg_basebackup/receivelog.h
+1-1Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,6 @@ extern bool ReceiveXlogStream(PGconn *conn,
3131
stream_stop_callback stream_stop,
3232
int standby_message_timeout,
3333
char *partial_suffix,
34-
int fsync_interval);
34+
bool synchronous);
3535

3636
#endif /* RECEIVELOG_H */

0 commit comments

Comments
0 (0)
Morty Proxy This is a proxified and sanitized view of the page, visit original site.