From fdca8988480cac602157c3ae24ae61311bdaf960 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 | 49 +++--------- src/bin/pg_basebackup/pg_recvlogical.c | 119 +++++---------------------- src/bin/pg_basebackup/streamutil.c | 142 +++++++++++++++++++++++++++++++++ src/bin/pg_basebackup/streamutil.h | 9 +++ 5 files changed, 183 insertions(+), 157 deletions(-) diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c index 8b9acea..49675cf 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)) 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..f722374 100644 --- a/src/bin/pg_basebackup/pg_receivexlog.c +++ b/src/bin/pg_basebackup/pg_receivexlog.c @@ -253,21 +253,10 @@ FindStreamingStart(uint32 *tli) static void StreamLog(void) { - PGresult *res; XLogRecPtr startpos; - uint32 starttli; + TimeLineID starttli; XLogRecPtr serverpos; - uint32 servertli; - uint32 hi, - lo; - - /* - * Connect in replication mode to the server - */ - conn = GetConnection(); - if (!conn) - /* Error message already written in GetConnection() */ - return; + TimeLineID servertli; if (!CheckServerVersionForStreaming(conn)) { @@ -280,33 +269,11 @@ 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. */ - 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)) 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. @@ -492,6 +459,12 @@ main(int argc, char **argv) pqsignal(SIGINT, sigint_handler); #endif + /* Obtain a connection before doing anything */ + conn = GetConnection(); + if (!conn) + /* Error message already written in GetConnection() */ + exit(1); + while (true) { StreamLog(); diff --git a/src/bin/pg_basebackup/pg_recvlogical.c b/src/bin/pg_basebackup/pg_recvlogical.c index f3b0f34..1e53c82 100644 --- a/src/bin/pg_basebackup/pg_recvlogical.c +++ b/src/bin/pg_basebackup/pg_recvlogical.c @@ -208,15 +208,6 @@ StreamLog(void) query = createPQExpBuffer(); /* - * Connect in replication mode to the server - */ - if (!conn) - conn = GetConnection(); - if (!conn) - /* Error message already written in GetConnection() */ - return; - - /* * Start the replication */ if (verbose) @@ -596,7 +587,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'}, @@ -834,121 +824,48 @@ 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); - /* - * stop a replication slot - */ + /* Drop a replication slot */ if (do_drop_slot) { - char query[256]; - 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)); - 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); + if (!DropReplicationSlot(conn, false)) 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, 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); + /* Identify system, obtaining start LSN position at the same time */ + if (!RunIdentifySystem(conn, NULL, NULL, &startpos)) + disconnect_and_exit(1); + + /* Stream loop */ while (true) { StreamLog(); diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c index 1100260..173dda4 100644 --- a/src/bin/pg_basebackup/streamutil.c +++ b/src/bin/pg_basebackup/streamutil.c @@ -227,6 +227,148 @@ 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 + */ +bool +RunIdentifySystem(PGconn *conn, char **sysid, TimeLineID *starttli, + XLogRecPtr *startpos) +{ + PGresult *res; + uint32 hi, lo; + + /* Leave if no connection present */ + if (conn == NULL) + return false; + + 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, required %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 (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; + } + if (startpos != NULL) + *startpos = ((uint64) hi) << 32 | lo; + + 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, + bool is_physical) +{ + char query[256]; + PGresult *res; + uint32 hi, lo; + + /* 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; + } + + /* Check LSN format obtained as consistent point */ + 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; + } + + 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; + + /* 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..e03687c 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,13 @@ extern PGconn *conn; extern PGconn *GetConnection(void); +/* Replication commands */ +extern bool CreateReplicationSlot(PGconn *conn, const char *plugin, + bool is_physical); +extern bool DropReplicationSlot(PGconn *conn, bool is_physical); +extern bool RunIdentifySystem(PGconn *conn, char **sysid, + TimeLineID *starttli, + XLogRecPtr *startpos); extern int64 feGetCurrentTimestamp(void); extern void feTimestampDifference(int64 start_time, int64 stop_time, long *secs, int *microsecs); -- 2.1.0