From 6f237d4ad74c1e9c699e6e4302afb4f75b8e16b9 Mon Sep 17 00:00:00 2001 From: Michael Paquier Date: Mon, 1 Sep 2014 20:48:43 +0900 Subject: [PATCH 1/2] Refactoring of pg_basebackup utilities Code duplication is reduced with the introduction of new APIs for each individual replication command: - IDENTIFY_SYSTEM - CREATE_REPLICATION_SLOT - DROP_REPLICATION_SLOT A couple of variables used to identify a timeline ID are changed as well to be more consistent with core code. --- src/bin/pg_basebackup/pg_basebackup.c | 21 +---- src/bin/pg_basebackup/pg_receivexlog.c | 38 ++------ src/bin/pg_basebackup/pg_recvlogical.c | 120 +++++++------------------ src/bin/pg_basebackup/streamutil.c | 159 +++++++++++++++++++++++++++++++++ src/bin/pg_basebackup/streamutil.h | 10 +++ 5 files changed, 209 insertions(+), 139 deletions(-) diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c index 8b9acea..0ebda9a 100644 --- a/src/bin/pg_basebackup/pg_basebackup.c +++ b/src/bin/pg_basebackup/pg_basebackup.c @@ -1569,8 +1569,8 @@ BaseBackup(void) { PGresult *res; char *sysidentifier; - uint32 latesttli; - uint32 starttli; + TimeLineID latesttli; + TimeLineID starttli; char *basebkp; char escaped_label[MAXPGPATH]; char *maxrate_clause = NULL; @@ -1624,23 +1624,8 @@ BaseBackup(void) /* * Run IDENTIFY_SYSTEM so we can get the timeline */ - res = PQexec(conn, "IDENTIFY_SYSTEM"); - if (PQresultStatus(res) != PGRES_TUPLES_OK) - { - fprintf(stderr, _("%s: could not send replication command \"%s\": %s"), - progname, "IDENTIFY_SYSTEM", PQerrorMessage(conn)); + if (!RunIdentifySystem(conn, &sysidentifier, &latesttli, NULL, NULL)) disconnect_and_exit(1); - } - if (PQntuples(res) != 1 || PQnfields(res) < 3) - { - fprintf(stderr, - _("%s: could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields\n"), - progname, PQntuples(res), PQnfields(res), 1, 3); - disconnect_and_exit(1); - } - sysidentifier = pg_strdup(PQgetvalue(res, 0, 0)); - latesttli = atoi(PQgetvalue(res, 0, 1)); - PQclear(res); /* * Start the actual backup diff --git a/src/bin/pg_basebackup/pg_receivexlog.c b/src/bin/pg_basebackup/pg_receivexlog.c index a8b9ad3..171cf43 100644 --- a/src/bin/pg_basebackup/pg_receivexlog.c +++ b/src/bin/pg_basebackup/pg_receivexlog.c @@ -253,13 +253,8 @@ FindStreamingStart(uint32 *tli) static void StreamLog(void) { - PGresult *res; - XLogRecPtr startpos; - uint32 starttli; - XLogRecPtr serverpos; - uint32 servertli; - uint32 hi, - lo; + XLogRecPtr startpos, serverpos; + TimeLineID starttli, servertli; /* * Connect in replication mode to the server @@ -280,33 +275,12 @@ StreamLog(void) } /* - * Run IDENTIFY_SYSTEM so we can get the timeline and current xlog - * position. + * Identify server, obtaining start LSN position and current timeline ID + * at the same time, necessary if not valid data can be found in the + * existing output directory. */ - res = PQexec(conn, "IDENTIFY_SYSTEM"); - if (PQresultStatus(res) != PGRES_TUPLES_OK) - { - fprintf(stderr, _("%s: could not send replication command \"%s\": %s"), - progname, "IDENTIFY_SYSTEM", PQerrorMessage(conn)); - disconnect_and_exit(1); - } - if (PQntuples(res) != 1 || PQnfields(res) < 3) - { - fprintf(stderr, - _("%s: could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields\n"), - progname, PQntuples(res), PQnfields(res), 1, 3); + if (!RunIdentifySystem(conn, NULL, &servertli, &serverpos, NULL)) disconnect_and_exit(1); - } - servertli = atoi(PQgetvalue(res, 0, 1)); - if (sscanf(PQgetvalue(res, 0, 2), "%X/%X", &hi, &lo) != 2) - { - fprintf(stderr, - _("%s: could not parse transaction log location \"%s\"\n"), - progname, PQgetvalue(res, 0, 2)); - disconnect_and_exit(1); - } - serverpos = ((uint64) hi) << 32 | lo; - PQclear(res); /* * Figure out where to start streaming. diff --git a/src/bin/pg_basebackup/pg_recvlogical.c b/src/bin/pg_basebackup/pg_recvlogical.c index f3b0f34..23ae225 100644 --- a/src/bin/pg_basebackup/pg_recvlogical.c +++ b/src/bin/pg_basebackup/pg_recvlogical.c @@ -596,7 +596,6 @@ sighup_handler(int signum) int main(int argc, char **argv) { - PGresult *res; static struct option long_options[] = { /* general options */ {"file", required_argument, NULL, 'f'}, @@ -628,6 +627,7 @@ main(int argc, char **argv) int option_index; uint32 hi, lo; + char *db_name; progname = get_progname(argv[0]); set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_recvlogical")); @@ -834,121 +834,63 @@ main(int argc, char **argv) #endif /* - * don't really need this but it actually helps to get more precise error - * messages about authentication, required GUCs and such without starting - * to loop around connection attempts lateron. + * Obtain a connection to server. This is not really necessary but it + * helps to get more precise error messages about authentification, + * required GUC parameters and such. */ - { - conn = GetConnection(); - if (!conn) - /* Error message already written in GetConnection() */ - exit(1); - - /* - * Run IDENTIFY_SYSTEM so we can get the timeline and current xlog - * position. - */ - res = PQexec(conn, "IDENTIFY_SYSTEM"); - if (PQresultStatus(res) != PGRES_TUPLES_OK) - { - fprintf(stderr, _("%s: could not send replication command \"%s\": %s"), - progname, "IDENTIFY_SYSTEM", PQerrorMessage(conn)); - disconnect_and_exit(1); - } - - if (PQntuples(res) != 1 || PQnfields(res) < 4) - { - fprintf(stderr, - _("%s: could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields\n"), - progname, PQntuples(res), PQnfields(res), 1, 4); - disconnect_and_exit(1); - } - PQclear(res); - } + conn = GetConnection(); + if (!conn) + /* Error message already written in GetConnection() */ + exit(1); + /* + * Run IDENTIFY_SYSTEM so we can get the timeline and current xlog + * position. + */ + if (!RunIdentifySystem(conn, NULL, NULL, NULL, &db_name)) + disconnect_and_exit(1); /* - * stop a replication slot + * Check that there is a database associated with connection, one is + * needed in this context. */ - if (do_drop_slot) + if (!db_name) { - char query[256]; + fprintf(stderr, + _("%s: no database defined for replication connection \"%s\"\n"), + progname, replication_slot); + disconnect_and_exit(1); + } + /* Drop a replication slot */ + if (do_drop_slot) + { if (verbose) fprintf(stderr, - _("%s: freeing replication slot \"%s\"\n"), + _("%s: dropping replication slot \"%s\"\n"), progname, replication_slot); - snprintf(query, sizeof(query), "DROP_REPLICATION_SLOT \"%s\"", - replication_slot); - res = PQexec(conn, query); - if (PQresultStatus(res) != PGRES_COMMAND_OK) - { - fprintf(stderr, _("%s: could not send replication command \"%s\": %s"), - progname, query, PQerrorMessage(conn)); + if (!DropReplicationSlot(conn, false)) disconnect_and_exit(1); - } - - if (PQntuples(res) != 0 || PQnfields(res) != 0) - { - fprintf(stderr, - _("%s: could not stop logical replication: got %d rows and %d fields, expected %d rows and %d fields\n"), - progname, PQntuples(res), PQnfields(res), 0, 0); - disconnect_and_exit(1); - } - - PQclear(res); disconnect_and_exit(0); } - /* - * init a replication slot - */ + /* Create a replication slot */ if (do_create_slot) { - char query[256]; - if (verbose) fprintf(stderr, - _("%s: initializing replication slot \"%s\"\n"), + _("%s: creating replication slot \"%s\"\n"), progname, replication_slot); - snprintf(query, sizeof(query), "CREATE_REPLICATION_SLOT \"%s\" LOGICAL \"%s\"", - replication_slot, plugin); - - res = PQexec(conn, query); - if (PQresultStatus(res) != PGRES_TUPLES_OK) - { - fprintf(stderr, _("%s: could not send replication command \"%s\": %s"), - progname, query, PQerrorMessage(conn)); + if (!CreateReplicationSlot(conn, plugin, &startpos, false)) disconnect_and_exit(1); - } - - if (PQntuples(res) != 1 || PQnfields(res) != 4) - { - fprintf(stderr, - _("%s: could not init logical replication: got %d rows and %d fields, expected %d rows and %d fields\n"), - progname, PQntuples(res), PQnfields(res), 1, 4); - disconnect_and_exit(1); - } - - if (sscanf(PQgetvalue(res, 0, 1), "%X/%X", &hi, &lo) != 2) - { - fprintf(stderr, - _("%s: could not parse transaction log location \"%s\"\n"), - progname, PQgetvalue(res, 0, 1)); - disconnect_and_exit(1); - } - startpos = ((uint64) hi) << 32 | lo; - - replication_slot = strdup(PQgetvalue(res, 0, 0)); - PQclear(res); } - if (!do_start_slot) disconnect_and_exit(0); + /* Stream loop */ while (true) { StreamLog(); diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c index 1100260..2c373b8 100644 --- a/src/bin/pg_basebackup/streamutil.c +++ b/src/bin/pg_basebackup/streamutil.c @@ -227,6 +227,165 @@ GetConnection(void) return tmpconn; } +/* + * Run IDENTIFY_SYSTEM through a given connection and give back to caller + * some result information if requested: + * - Start LSN position + * - Current timeline ID + * - System identifier + * - Plugin name + */ +bool +RunIdentifySystem(PGconn *conn, char **sysid, TimeLineID *starttli, + XLogRecPtr *startpos, char **db_name) +{ + PGresult *res; + uint32 hi, lo; + + /* Check connection existence */ + Assert(conn != NULL); + + res = PQexec(conn, "IDENTIFY_SYSTEM"); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + fprintf(stderr, _("%s: could not send replication command \"%s\": %s"), + progname, "IDENTIFY_SYSTEM", PQerrorMessage(conn)); + return false; + } + if (PQntuples(res) != 1 || PQnfields(res) < 3) + { + fprintf(stderr, + _("%s: could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields\n"), + progname, PQntuples(res), PQnfields(res), 1, 3); + return false; + } + + /* Get system identifier */ + if (sysid != NULL) + *sysid = pg_strdup(PQgetvalue(res, 0, 0)); + + /* Get timeline ID to start streaming from */ + if (starttli != NULL) + *starttli = atoi(PQgetvalue(res, 0, 1)); + + /* Get LSN start position if necessary */ + if (startpos != NULL) + { + if (sscanf(PQgetvalue(res, 0, 2), "%X/%X", &hi, &lo) != 2) + { + fprintf(stderr, + _("%s: could not parse transaction log location \"%s\"\n"), + progname, PQgetvalue(res, 0, 2)); + return false; + } + *startpos = ((uint64) hi) << 32 | lo; + } + + /* Get database name, only available in 9.4 and newer versions */ + if (db_name != NULL) + *db_name = PQnfields(res) > 3 && !PQgetisnull(res, 0, 3) ? + pg_strdup(PQgetvalue(res, 0, 3)) : (char *) NULL; + + PQclear(res); + return true; +} + +/* + * Create a replication slot for the given connection. This function + * returns true in case of success as well as the start position + * obtained after the slot creation. + */ +bool +CreateReplicationSlot(PGconn *conn, const char *plugin, + XLogRecPtr *startpos, bool is_physical) +{ + char query[256]; + PGresult *res; + uint32 hi, lo; + + Assert((is_physical && plugin == NULL) || + (!is_physical && plugin != NULL)); + Assert(replication_slot != NULL); + + /* Build query */ + if (is_physical) + snprintf(query, sizeof(query), "CREATE_REPLICATION_SLOT \"%s\" PHYSICAL", + replication_slot); + else + snprintf(query, sizeof(query), "CREATE_REPLICATION_SLOT \"%s\" LOGICAL \"%s\"", + replication_slot, plugin); + + res = PQexec(conn, query); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + fprintf(stderr, _("%s: could not send replication command \"%s\": %s"), + progname, query, PQerrorMessage(conn)); + return false; + } + + if (PQntuples(res) != 1 || PQnfields(res) != 4) + { + fprintf(stderr, + _("%s: could not init %s replication: got %d rows and %d fields, expected %d rows and %d fields\n"), + progname, is_physical ? "physical" : "logical", + PQntuples(res), PQnfields(res), 1, 4); + return false; + } + + /* Get LSN start position if necessary */ + if (startpos != NULL) + { + if (sscanf(PQgetvalue(res, 0, 1), "%X/%X", &hi, &lo) != 2) + { + fprintf(stderr, + _("%s: could not parse transaction log location \"%s\"\n"), + progname, PQgetvalue(res, 0, 1)); + return false; + } + *startpos = ((uint64) hi) << 32 | lo; + } + + replication_slot = strdup(PQgetvalue(res, 0, 0)); + PQclear(res); + return true; +} + +/* + * Drop a replication slot for the given connection. This function + * returns true in case of success. + */ +bool +DropReplicationSlot(PGconn *conn, bool is_physical) +{ + char query[256]; + PGresult *res; + + Assert(replication_slot != NULL); + + /* Build query */ + snprintf(query, sizeof(query), "DROP_REPLICATION_SLOT \"%s\"", + replication_slot); + res = PQexec(conn, query); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + fprintf(stderr, _("%s: could not send replication command \"%s\": %s"), + progname, query, PQerrorMessage(conn)); + return false; + } + + if (PQntuples(res) != 0 || PQnfields(res) != 0) + { + fprintf(stderr, + _("%s: could not stop %s replication: got %d rows and %d fields, expected %d rows and %d fields\n"), + progname, is_physical ? "physical" : "logical", + PQntuples(res), PQnfields(res), 0, 0); + return false; + } + + PQclear(res); + return true; +} + /* * Frontend version of GetCurrentTimestamp(), since we are not linked with diff --git a/src/bin/pg_basebackup/streamutil.h b/src/bin/pg_basebackup/streamutil.h index 8c6691f..9067810 100644 --- a/src/bin/pg_basebackup/streamutil.h +++ b/src/bin/pg_basebackup/streamutil.h @@ -14,6 +14,8 @@ #include "libpq-fe.h" +#include "access/xlogdefs.h" + extern const char *progname; extern char *connection_string; extern char *dbhost; @@ -28,6 +30,14 @@ extern PGconn *conn; extern PGconn *GetConnection(void); +/* Replication commands */ +extern bool CreateReplicationSlot(PGconn *conn, const char *plugin, + XLogRecPtr *startpos, bool is_physical); +extern bool DropReplicationSlot(PGconn *conn, bool is_physical); +extern bool RunIdentifySystem(PGconn *conn, char **sysid, + TimeLineID *starttli, + XLogRecPtr *startpos, + char **db_name); extern int64 feGetCurrentTimestamp(void); extern void feTimestampDifference(int64 start_time, int64 stop_time, long *secs, int *microsecs); -- 2.1.0