31 #define RECONNECT_SLEEP_TIME 5 49 static void usage(
void);
55 bool segment_finished);
65 #define IsCompressXLogFileName(fname) \ 66 (strlen(fname) == XLOG_FNAME_LEN + strlen(".gz") && \ 67 strspn(fname, "0123456789ABCDEF") == XLOG_FNAME_LEN && \ 68 strcmp((fname) + XLOG_FNAME_LEN, ".gz") == 0) 69 #define IsPartialCompressXLogFileName(fname) \ 70 (strlen(fname) == XLOG_FNAME_LEN + strlen(".gz.partial") && \ 71 strspn(fname, "0123456789ABCDEF") == XLOG_FNAME_LEN && \ 72 strcmp((fname) + XLOG_FNAME_LEN, ".gz.partial") == 0) 77 printf(
_(
"%s receives PostgreSQL streaming write-ahead logs.\n\n"),
82 printf(
_(
" -D, --directory=DIR receive write-ahead log files into this directory\n"));
83 printf(
_(
" -E, --endpos=LSN exit after receiving the specified LSN\n"));
84 printf(
_(
" --if-not-exists do not error if slot already exists when creating a slot\n"));
85 printf(
_(
" -n, --no-loop do not loop on connection lost\n"));
86 printf(
_(
" --no-sync do not wait for changes to be written safely to disk\n"));
87 printf(
_(
" -s, --status-interval=SECS\n" 89 printf(
_(
" -S, --slot=SLOTNAME replication slot to use\n"));
90 printf(
_(
" --synchronous flush write-ahead log immediately after writing\n"));
91 printf(
_(
" -v, --verbose output verbose messages\n"));
92 printf(
_(
" -V, --version output version information, then exit\n"));
93 printf(
_(
" -Z, --compress=0-9 compress logs with given compression level\n"));
94 printf(
_(
" -?, --help show this help, then exit\n"));
95 printf(
_(
"\nConnection options:\n"));
96 printf(
_(
" -d, --dbname=CONNSTR connection string\n"));
97 printf(
_(
" -h, --host=HOSTNAME database server host or socket directory\n"));
98 printf(
_(
" -p, --port=PORT database server port number\n"));
99 printf(
_(
" -U, --username=NAME connect as specified database user\n"));
100 printf(
_(
" -w, --no-password never prompt for password\n"));
101 printf(
_(
" -W, --password force password prompt (should happen automatically)\n"));
102 printf(
_(
"\nOptional actions:\n"));
103 printf(
_(
" --create-slot create a new replication slot (for the slot's name see --slot)\n"));
104 printf(
_(
" --drop-slot drop the replication slot (for the slot's name see --slot)\n"));
105 printf(
_(
"\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT);
106 printf(
_(
"%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL);
112 static uint32 prevtimeline = 0;
116 if (
verbose && segment_finished)
117 pg_log_info(
"finished segment at %X/%X (timeline %u)",
124 pg_log_info(
"stopped log streaming at %X/%X (timeline %u)",
139 if (
verbose && prevtimeline != 0 && prevtimeline != timeline)
144 prevtimeline = timeline;
165 Assert(dest_folder != NULL);
183 Assert(dest_dir != NULL && dest_folder != NULL);
186 pg_log_error(
"could not close directory \"%s\": %m", dest_folder);
206 bool high_ispartial =
false;
210 while (errno = 0, (dirent =
readdir(dir)) != NULL)
258 if (!ispartial && !iscompress)
264 if (
stat(fullpath, &statbuf) != 0)
266 pg_log_error(
"could not stat file \"%s\": %m", fullpath);
272 pg_log_warning(
"segment file \"%s\" has incorrect size %lld, skipping",
277 else if (!ispartial && iscompress)
287 fd = open(fullpath, O_RDONLY |
PG_BINARY, 0);
290 pg_log_error(
"could not open compressed file \"%s\": %m",
294 if (lseek(fd, (off_t) (-4), SEEK_END) < 0)
296 pg_log_error(
"could not seek in compressed file \"%s\": %m",
300 r =
read(fd, (
char *) buf,
sizeof(buf));
301 if (r !=
sizeof(buf))
304 pg_log_error(
"could not read compressed file \"%s\": %m",
307 pg_log_error(
"could not read compressed file \"%s\": read %d of %zu",
308 fullpath, r,
sizeof(buf));
313 bytes_out = (buf[3] << 24) | (buf[2] << 16) |
314 (buf[1] << 8) | buf[0];
318 pg_log_warning(
"compressed segment file \"%s\" has incorrect uncompressed size %d, skipping",
319 dirent->
d_name, bytes_out);
325 if ((segno > high_segno) ||
326 (segno == high_segno && tli > high_tli) ||
327 (segno == high_segno && tli == high_tli && high_ispartial && !ispartial))
331 high_ispartial = ispartial;
374 MemSet(&stream, 0,
sizeof(stream));
422 pg_log_info(
"starting log streaming at %X/%X (timeline %u)",
441 pg_log_info(
"could not finish writing WAL files: %m");
470 static struct option long_options[] = {
507 if (strcmp(argv[1],
"--help") == 0 || strcmp(argv[1],
"-?") == 0)
512 else if (strcmp(argv[1],
"-V") == 0 ||
513 strcmp(argv[1],
"--version") == 0)
515 puts(
"pg_receivewal (PostgreSQL) " PG_VERSION);
520 while ((c =
getopt_long(argc, argv,
"D:d:E:h:p:U:s:S:nwWvZ:",
521 long_options, &option_index)) != -1)
563 if (sscanf(
optarg,
"%X/%X", &hi, &lo) != 2)
568 endpos = ((uint64) hi) << 32 | lo;
578 if (compresslevel < 0 || compresslevel > 9)
605 fprintf(stderr,
_(
"Try \"%s --help\" for more information.\n"),
616 pg_log_error(
"too many command-line arguments (first is \"%s\")",
618 fprintf(stderr,
_(
"Try \"%s --help\" for more information.\n"),
625 pg_log_error(
"cannot use --create-slot together with --drop-slot");
626 fprintf(stderr,
_(
"Try \"%s --help\" for more information.\n"),
634 pg_log_error(
"%s needs a slot to be specified using --slot",
636 fprintf(stderr,
_(
"Try \"%s --help\" for more information.\n"),
643 pg_log_error(
"cannot use --synchronous together with --no-sync");
644 fprintf(stderr,
_(
"Try \"%s --help\" for more information.\n"),
655 fprintf(stderr,
_(
"Try \"%s --help\" for more information.\n"),
663 pg_log_error(
"this build does not support compression");
719 pg_log_error(
"replication connection using slot \"%s\" is unexpectedly database specific",
773 pg_log_info(
"disconnected; waiting %d seconds to try again",
static char * replication_slot
#define InvalidXLogRecPtr
int main(int argc, char **argv)
static int standby_message_timeout
const char * get_progname(const char *argv0)
static void DropReplicationSlot(DropReplicationSlotCmd *cmd)
#define pg_log_error(...)
int getopt_long(int argc, char *const argv[], const char *optstring, const struct option *longopts, int *longindex)
void pg_logging_init(const char *argv0)
#define IsPartialCompressXLogFileName(fname)
bool RunIdentifySystem(PGconn *conn, char **sysid, TimeLineID *starttli, XLogRecPtr *startpos, char **db_name)
void PQfinish(PGconn *conn)
#define MemSet(start, val, len)
static bool stop_streaming(XLogRecPtr segendpos, uint32 timeline, bool segment_finished)
static void close_destination_dir(DIR *dest_dir, char *dest_folder)
static int fd(const char *x, int i)
WalWriteMethod * CreateWalDirectoryMethod(const char *basedir, int compression, bool sync)
static void sigint_handler(int signum)
static bool slot_exists_ok
static void StreamLog(void)
void pg_usleep(long microsec)
#define LSN_FORMAT_ARGS(lsn)
#define required_argument
#define IsXLogFileName(fname)
bool RetrieveWalSegSize(PGconn *conn)
DIR * opendir(const char *)
#define XLogFromFileName(fname, tli, logSegNo, wal_segsz_bytes)
static XLogRecPtr FindStreamingStart(uint32 *tli)
char * pg_strdup(const char *in)
static volatile bool time_to_stop
stream_stop_callback stream_stop
WalWriteMethod * walmethod
#define XLogRecPtrIsInvalid(r)
#define PG_TEXTDOMAIN(domain)
#define XLogSegmentOffset(xlogptr, wal_segsz_bytes)
static void disconnect_atexit(void)
pqsigfunc pqsignal(int signum, pqsigfunc handler)
static DIR * get_destination_dir(char *dest_folder)
PGconn * GetConnection(UserMapping *user, bool will_prep_stmt)
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
#define Assert(condition)
bool ReceiveXlogStream(PGconn *conn, StreamCtl *stream)
#define RECONNECT_SLEEP_TIME
struct dirent * readdir(DIR *)
static bool do_create_slot
static void CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
#define IsPartialXLogFileName(fname)
void set_pglocale_pgservice(const char *argv0, const char *app)
int standby_message_timeout
#define pg_log_warning(...)
#define IsCompressXLogFileName(fname)
void FreeWalDirectoryMethod(void)
bool CheckServerVersionForStreaming(PGconn *conn)