PostgreSQL Source Code git master
All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Pages
receivelog.h File Reference
#include "access/xlogdefs.h"
#include "libpq-fe.h"
#include "walmethods.h"
Include dependency graph for receivelog.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures

struct  StreamCtl
 

Typedefs

typedef bool(* stream_stop_callback) (XLogRecPtr segendpos, uint32 timeline, bool segment_finished)
 
typedef struct StreamCtl StreamCtl
 

Functions

bool CheckServerVersionForStreaming (PGconn *conn)
 
bool ReceiveXlogStream (PGconn *conn, StreamCtl *stream)
 

Typedef Documentation

◆ stream_stop_callback

typedef bool(* stream_stop_callback) (XLogRecPtr segendpos, uint32 timeline, bool segment_finished)

Definition at line 23 of file receivelog.h.

◆ StreamCtl

typedef struct StreamCtl StreamCtl

Function Documentation

◆ CheckServerVersionForStreaming()

bool CheckServerVersionForStreaming ( PGconn conn)

Definition at line 374 of file receivelog.c.

375{
376 int minServerMajor,
377 maxServerMajor;
378 int serverMajor;
379
380 /*
381 * The message format used in streaming replication changed in 9.3, so we
382 * cannot stream from older servers. And we don't support servers newer
383 * than the client; it might work, but we don't know, so err on the safe
384 * side.
385 */
386 minServerMajor = 903;
387 maxServerMajor = PG_VERSION_NUM / 100;
388 serverMajor = PQserverVersion(conn) / 100;
389 if (serverMajor < minServerMajor)
390 {
391 const char *serverver = PQparameterStatus(conn, "server_version");
392
393 pg_log_error("incompatible server version %s; client does not support streaming from server versions older than %s",
394 serverver ? serverver : "'unknown'",
395 "9.3");
396 return false;
397 }
398 else if (serverMajor > maxServerMajor)
399 {
400 const char *serverver = PQparameterStatus(conn, "server_version");
401
402 pg_log_error("incompatible server version %s; client does not support streaming from server versions newer than %s",
403 serverver ? serverver : "'unknown'",
404 PG_VERSION);
405 return false;
406 }
407 return true;
408}
int PQserverVersion(const PGconn *conn)
Definition: fe-connect.c:7199
const char * PQparameterStatus(const PGconn *conn, const char *paramName)
Definition: fe-connect.c:7164
#define pg_log_error(...)
Definition: logging.h:106
PGconn * conn
Definition: streamutil.c:53

References conn, pg_log_error, PQparameterStatus(), and PQserverVersion().

Referenced by BaseBackup(), ReceiveXlogStream(), and StreamLog().

◆ ReceiveXlogStream()

bool ReceiveXlogStream ( PGconn conn,
StreamCtl stream 
)

Definition at line 452 of file receivelog.c.

453{
454 char query[128];
455 char slotcmd[128];
456 PGresult *res;
457 XLogRecPtr stoppos;
458
459 /*
460 * The caller should've checked the server version already, but doesn't do
461 * any harm to check it here too.
462 */
464 return false;
465
466 /*
467 * Decide whether we want to report the flush position. If we report the
468 * flush position, the primary will know what WAL we'll possibly
469 * re-request, and it can then remove older WAL safely. We must always do
470 * that when we are using slots.
471 *
472 * Reporting the flush position makes one eligible as a synchronous
473 * replica. People shouldn't include generic names in
474 * synchronous_standby_names, but we've protected them against it so far,
475 * so let's continue to do so unless specifically requested.
476 */
477 if (stream->replication_slot != NULL)
478 {
479 reportFlushPosition = true;
480 sprintf(slotcmd, "SLOT \"%s\" ", stream->replication_slot);
481 }
482 else
483 {
484 if (stream->synchronous)
485 reportFlushPosition = true;
486 else
487 reportFlushPosition = false;
488 slotcmd[0] = 0;
489 }
490
491 if (stream->sysidentifier != NULL)
492 {
493 char *sysidentifier = NULL;
494 TimeLineID servertli;
495
496 /*
497 * Get the server system identifier and timeline, and validate them.
498 */
499 if (!RunIdentifySystem(conn, &sysidentifier, &servertli, NULL, NULL))
500 {
501 pg_free(sysidentifier);
502 return false;
503 }
504
505 if (strcmp(stream->sysidentifier, sysidentifier) != 0)
506 {
507 pg_log_error("system identifier does not match between base backup and streaming connection");
508 pg_free(sysidentifier);
509 return false;
510 }
511 pg_free(sysidentifier);
512
513 if (stream->timeline > servertli)
514 {
515 pg_log_error("starting timeline %u is not present in the server",
516 stream->timeline);
517 return false;
518 }
519 }
520
521 /*
522 * initialize flush position to starting point, it's the caller's
523 * responsibility that that's sane.
524 */
525 lastFlushPosition = stream->startpos;
526
527 while (1)
528 {
529 /*
530 * Fetch the timeline history file for this timeline, if we don't have
531 * it already. When streaming log to tar, this will always return
532 * false, as we are never streaming into an existing file and
533 * therefore there can be no pre-existing timeline history file.
534 */
535 if (!existsTimeLineHistoryFile(stream))
536 {
537 snprintf(query, sizeof(query), "TIMELINE_HISTORY %u", stream->timeline);
538 res = PQexec(conn, query);
540 {
541 /* FIXME: we might send it ok, but get an error */
542 pg_log_error("could not send replication command \"%s\": %s",
543 "TIMELINE_HISTORY", PQresultErrorMessage(res));
544 PQclear(res);
545 return false;
546 }
547
548 /*
549 * The response to TIMELINE_HISTORY is a single row result set
550 * with two fields: filename and content
551 */
552 if (PQnfields(res) != 2 || PQntuples(res) != 1)
553 {
554 pg_log_warning("unexpected response to TIMELINE_HISTORY command: got %d rows and %d fields, expected %d rows and %d fields",
555 PQntuples(res), PQnfields(res), 1, 2);
556 }
557
558 /* Write the history file to disk */
560 PQgetvalue(res, 0, 0),
561 PQgetvalue(res, 0, 1));
562
563 PQclear(res);
564 }
565
566 /*
567 * Before we start streaming from the requested location, check if the
568 * callback tells us to stop here.
569 */
570 if (stream->stream_stop(stream->startpos, stream->timeline, false))
571 return true;
572
573 /* Initiate the replication stream at specified location */
574 snprintf(query, sizeof(query), "START_REPLICATION %s%X/%X TIMELINE %u",
575 slotcmd,
576 LSN_FORMAT_ARGS(stream->startpos),
577 stream->timeline);
578 res = PQexec(conn, query);
580 {
581 pg_log_error("could not send replication command \"%s\": %s",
582 "START_REPLICATION", PQresultErrorMessage(res));
583 PQclear(res);
584 return false;
585 }
586 PQclear(res);
587
588 /* Stream the WAL */
589 res = HandleCopyStream(conn, stream, &stoppos);
590 if (res == NULL)
591 goto error;
592
593 /*
594 * Streaming finished.
595 *
596 * There are two possible reasons for that: a controlled shutdown, or
597 * we reached the end of the current timeline. In case of
598 * end-of-timeline, the server sends a result set after Copy has
599 * finished, containing information about the next timeline. Read
600 * that, and restart streaming from the next timeline. In case of
601 * controlled shutdown, stop here.
602 */
604 {
605 /*
606 * End-of-timeline. Read the next timeline's ID and starting
607 * position. Usually, the starting position will match the end of
608 * the previous timeline, but there are corner cases like if the
609 * server had sent us half of a WAL record, when it was promoted.
610 * The new timeline will begin at the end of the last complete
611 * record in that case, overlapping the partial WAL record on the
612 * old timeline.
613 */
614 uint32 newtimeline;
615 bool parsed;
616
617 parsed = ReadEndOfStreamingResult(res, &stream->startpos, &newtimeline);
618 PQclear(res);
619 if (!parsed)
620 goto error;
621
622 /* Sanity check the values the server gave us */
623 if (newtimeline <= stream->timeline)
624 {
625 pg_log_error("server reported unexpected next timeline %u, following timeline %u",
626 newtimeline, stream->timeline);
627 goto error;
628 }
629 if (stream->startpos > stoppos)
630 {
631 pg_log_error("server stopped streaming timeline %u at %X/%X, but reported next timeline %u to begin at %X/%X",
632 stream->timeline, LSN_FORMAT_ARGS(stoppos),
633 newtimeline, LSN_FORMAT_ARGS(stream->startpos));
634 goto error;
635 }
636
637 /* Read the final result, which should be CommandComplete. */
640 {
641 pg_log_error("unexpected termination of replication stream: %s",
643 PQclear(res);
644 goto error;
645 }
646 PQclear(res);
647
648 /*
649 * Loop back to start streaming from the new timeline. Always
650 * start streaming at the beginning of a segment.
651 */
652 stream->timeline = newtimeline;
653 stream->startpos = stream->startpos -
655 continue;
656 }
658 {
659 PQclear(res);
660
661 /*
662 * End of replication (ie. controlled shut down of the server).
663 *
664 * Check if the callback thinks it's OK to stop here. If not,
665 * complain.
666 */
667 if (stream->stream_stop(stoppos, stream->timeline, false))
668 return true;
669 else
670 {
671 pg_log_error("replication stream was terminated before stop point");
672 goto error;
673 }
674 }
675 else
676 {
677 /* Server returned an error. */
678 pg_log_error("unexpected termination of replication stream: %s",
680 PQclear(res);
681 goto error;
682 }
683 }
684
685error:
686 if (walfile != NULL && stream->walmethod->ops->close(walfile, CLOSE_NO_RENAME) != 0)
687 pg_log_error("could not close file \"%s\": %s",
689 walfile = NULL;
690 return false;
691}
uint32_t uint32
Definition: c.h:485
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3876
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:2062
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:3411
int PQntuples(const PGresult *res)
Definition: fe-exec.c:3481
char * PQresultErrorMessage(const PGresult *res)
Definition: fe-exec.c:3427
PGresult * PQexec(PGconn *conn, const char *query)
Definition: fe-exec.c:2262
int PQnfields(const PGresult *res)
Definition: fe-exec.c:3489
void pg_free(void *ptr)
Definition: fe_memutils.c:105
@ PGRES_COPY_BOTH
Definition: libpq-fe.h:132
@ PGRES_COMMAND_OK
Definition: libpq-fe.h:120
@ PGRES_TUPLES_OK
Definition: libpq-fe.h:123
#define pg_log_warning(...)
Definition: pgfnames.c:24
#define sprintf
Definition: port.h:240
#define snprintf
Definition: port.h:238
static PGresult * HandleCopyStream(PGconn *conn, StreamCtl *stream, XLogRecPtr *stoppos)
Definition: receivelog.c:744
static bool reportFlushPosition
Definition: receivelog.c:29
static bool existsTimeLineHistoryFile(StreamCtl *stream)
Definition: receivelog.c:257
static bool writeTimeLineHistoryFile(StreamCtl *stream, char *filename, char *content)
Definition: receivelog.c:274
static Walfile * walfile
Definition: receivelog.c:28
bool CheckServerVersionForStreaming(PGconn *conn)
Definition: receivelog.c:374
static XLogRecPtr lastFlushPosition
Definition: receivelog.c:30
static bool ReadEndOfStreamingResult(PGresult *res, XLogRecPtr *startpos, uint32 *timeline)
Definition: receivelog.c:698
static void error(void)
Definition: sql-dyntest.c:147
int WalSegSz
Definition: streamutil.c:32
bool RunIdentifySystem(PGconn *conn, char **sysid, TimeLineID *starttli, XLogRecPtr *startpos, char **db_name)
Definition: streamutil.c:478
char * sysidentifier
Definition: receivelog.h:33
TimeLineID timeline
Definition: receivelog.h:32
stream_stop_callback stream_stop
Definition: receivelog.h:41
char * replication_slot
Definition: receivelog.h:48
XLogRecPtr startpos
Definition: receivelog.h:31
WalWriteMethod * walmethod
Definition: receivelog.h:46
bool synchronous
Definition: receivelog.h:36
int(* close)(Walfile *f, WalCloseMethod method)
Definition: walmethods.h:55
const WalWriteMethodOps * ops
Definition: walmethods.h:105
char * pathname
Definition: walmethods.h:21
const char * GetLastWalMethodError(WalWriteMethod *wwmethod)
Definition: walmethods.c:1383
@ CLOSE_NO_RENAME
Definition: walmethods.h:35
#define XLogSegmentOffset(xlogptr, wal_segsz_bytes)
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
uint64 XLogRecPtr
Definition: xlogdefs.h:21
uint32 TimeLineID
Definition: xlogdefs.h:59

References CheckServerVersionForStreaming(), WalWriteMethodOps::close, CLOSE_NO_RENAME, conn, error(), existsTimeLineHistoryFile(), GetLastWalMethodError(), HandleCopyStream(), lastFlushPosition, LSN_FORMAT_ARGS, WalWriteMethod::ops, Walfile::pathname, pg_free(), pg_log_error, pg_log_warning, PGRES_COMMAND_OK, PGRES_COPY_BOTH, PGRES_TUPLES_OK, PQclear(), PQexec(), PQgetResult(), PQgetvalue(), PQnfields(), PQntuples(), PQresultErrorMessage(), PQresultStatus(), ReadEndOfStreamingResult(), StreamCtl::replication_slot, reportFlushPosition, res, RunIdentifySystem(), snprintf, sprintf, StreamCtl::startpos, StreamCtl::stream_stop, StreamCtl::synchronous, StreamCtl::sysidentifier, StreamCtl::timeline, walfile, StreamCtl::walmethod, WalSegSz, writeTimeLineHistoryFile(), and XLogSegmentOffset.

Referenced by LogStreamerMain(), and StreamLog().