#include "postgres_fe.h"
#include <sys/time.h>
#include <unistd.h>
#include "receivelog.h"
#include "streamutil.h"
#include "access/xlog_internal.h"
#include "common/fe_memutils.h"
#include "datatype/timestamp.h"
#include "port/pg_bswap.h"
#include "pqexpbuffer.h"

Include dependency graph for streamutil.c:

Macros
#define	ERRCODE_DUPLICATE_OBJECT "42710"

#define	MINIMUM_VERSION_FOR_SHOW_CMD 100000

Functions
PGconn *	GetConnection (void)

bool	RetrieveWalSegSize (PGconn *conn)

bool	RunIdentifySystem (PGconn conn, char sysid, TimeLineID starttli, XLogRecPtr startpos, char *db_name)

bool	CreateReplicationSlot (PGconn conn, const char slot_name, const char *plugin, bool is_temporary, bool is_physical, bool reserve_wal, bool slot_exists_ok)

bool	DropReplicationSlot (PGconn conn, const char slot_name)

TimestampTz	feGetCurrentTimestamp (void)

void	feTimestampDifference (TimestampTz start_time, TimestampTz stop_time, long secs, int microsecs)

bool	feTimestampDifferenceExceeds (TimestampTz start_time, TimestampTz stop_time, int msec)

void	fe_sendint64 (int64 i, char *buf)

int64	fe_recvint64 (char *buf)

Variables
uint32	WalSegSz

const char *	progname

char *	connection_string = NULL

char *	dbhost = NULL

char *	dbuser = NULL

char *	dbport = NULL

char *	dbname = NULL

int	dbgetpassword = 0

static bool	have_password = false

static char	password [100]

PGconn *	conn = NULL

Macro Definition Documentation

#define ERRCODE_DUPLICATE_OBJECT "42710"

Definition at line 30 of file streamutil.c.

Referenced by AddEnumLabel(), AddRelationNewConstraints(), AlterTypeNamespaceInternal(), ATExecAddConstraint(), CollationCreate(), ConversionCreate(), create_proc_lang(), CreateAccessMethod(), CreateCast(), CreateEventTrigger(), CreateExtension(), CreateForeignDataWrapper(), CreateForeignServer(), createNewConnection(), CreateOpFamily(), CreatePolicy(), CreatePublication(), CreateReplicationSlot(), CreateRole(), CreateStatistics(), CreateSubscription(), CreateTableSpace(), CreateTransform(), CreateTrigger(), CreateUserMapping(), DefineCompositeType(), DefineDomain(), DefineEnum(), DefineOpClass(), DefineRange(), DefineType(), domainAddConstraint(), get_domain_constraint_oid(), get_relation_constraint_oid(), GetDefaultOpClass(), heap_create_with_catalog(), InsertRule(), IsThereCollationInNamespace(), IsThereOpClassInNamespace(), IsThereOpFamilyInNamespace(), makeArrayTypeName(), MarkAsPreparing(), MergeCheckConstraint(), MergeWithExistingConstraint(), publication_add_relation(), RegisterExtensibleNodeEntry(), rename_policy(), RenameConstraintById(), RenameEnumLabel(), RenameRewriteRule(), RenameRole(), RenameTableSpace(), renametrig(), RenameTypeInternal(), ReplicationSlotCreate(), report_name_conflict(), report_namespace_conflict(), storeOperators(), storeProcedures(), transformGenericOptions(), and TypeCreate().

#define MINIMUM_VERSION_FOR_SHOW_CMD 100000

Definition at line 35 of file streamutil.c.

Referenced by RetrieveWalSegSize().

Function Documentation

bool CreateReplicationSlot	(	PGconn *	conn,
		const char *	slot_name,
		const char *	plugin,
		bool	is_temporary,
		bool	is_physical,
		bool	reserve_wal,
		bool	slot_exists_ok
	)

Definition at line 397 of file streamutil.c.

References _, appendPQExpBuffer(), Assert, createPQExpBuffer(), PQExpBufferData::data, destroyPQExpBuffer(), ERRCODE_DUPLICATE_OBJECT, PG_DIAG_SQLSTATE, PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQexec(), PQnfields(), PQntuples(), PQresultErrorField(), PQresultStatus(), PQserverVersion(), and progname.

 {
     PQExpBuffer query;
     PGresult   *res;
 
     query = createPQExpBuffer();
 
     Assert((is_physical && plugin == NULL) ||
            (!is_physical && plugin != NULL));
     Assert(slot_name != NULL);
 
     /* Build query */
     appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\"", slot_name);
     if (is_temporary)
         appendPQExpBuffer(query, " TEMPORARY");
     if (is_physical)
     {
         appendPQExpBuffer(query, " PHYSICAL");
         if (reserve_wal)
             appendPQExpBuffer(query, " RESERVE_WAL");
     }
     else
     {
         appendPQExpBuffer(query, " LOGICAL \"%s\"", plugin);
         if (PQserverVersion(conn) >= 100000)
             /* pg_recvlogical doesn't use an exported snapshot, so suppress */
             appendPQExpBuffer(query, " NOEXPORT_SNAPSHOT");
     }
 
     res = PQexec(conn, query->data);
     if (PQresultStatus(res) != PGRES_TUPLES_OK)
     {
         const char *sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
 
         if (slot_exists_ok &&
             sqlstate &&
             strcmp(sqlstate, ERRCODE_DUPLICATE_OBJECT) == 0)
         {
             destroyPQExpBuffer(query);
             PQclear(res);
             return true;
         }
         else
         {
             fprintf(stderr, _("%s: could not send replication command \"%s\": %s"),
                     progname, query->data, PQerrorMessage(conn));
 
             destroyPQExpBuffer(query);
             PQclear(res);
             return false;
         }
     }
 
     if (PQntuples(res) != 1 || PQnfields(res) != 4)
     {
         fprintf(stderr,
                 _("%s: could not create replication slot \"%s\": got %d rows and %d fields, expected %d rows and %d fields\n"),
                 progname, slot_name,
                 PQntuples(res), PQnfields(res), 1, 4);
 
         destroyPQExpBuffer(query);
         PQclear(res);
         return false;
     }
 
     destroyPQExpBuffer(query);
     PQclear(res);
     return true;
 }

bool DropReplicationSlot	(	PGconn *	conn,
		const char *	slot_name
	)

Definition at line 474 of file streamutil.c.

References _, appendPQExpBuffer(), Assert, createPQExpBuffer(), PQExpBufferData::data, destroyPQExpBuffer(), PGRES_COMMAND_OK, PQclear(), PQerrorMessage(), PQexec(), PQnfields(), PQntuples(), PQresultStatus(), and progname.

 {
     PQExpBuffer query;
     PGresult   *res;
 
     Assert(slot_name != NULL);
 
     query = createPQExpBuffer();
 
     /* Build query */
     appendPQExpBuffer(query, "DROP_REPLICATION_SLOT \"%s\"",
                       slot_name);
     res = PQexec(conn, query->data);
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
     {
         fprintf(stderr, _("%s: could not send replication command \"%s\": %s"),
                 progname, query->data, PQerrorMessage(conn));
 
         destroyPQExpBuffer(query);
         PQclear(res);
         return false;
     }
 
     if (PQntuples(res) != 0 || PQnfields(res) != 0)
     {
         fprintf(stderr,
                 _("%s: could not drop replication slot \"%s\": got %d rows and %d fields, expected %d rows and %d fields\n"),
                 progname, slot_name,
                 PQntuples(res), PQnfields(res), 0, 0);
 
         destroyPQExpBuffer(query);
         PQclear(res);
         return false;
     }
 
     destroyPQExpBuffer(query);
     PQclear(res);
     return true;
 }

int64 fe_recvint64 ( char * buf )

Definition at line 585 of file streamutil.c.

References pg_ntoh64.

Referenced by ProcessXLogDataMsg(), and StreamLogicalLog().

 {
     uint64      n64;
 
     memcpy(&n64, buf, sizeof(n64));
 
     return pg_ntoh64(n64);
 }

void fe_sendint64	(	int64	i,
		char *	buf
	)

Definition at line 574 of file streamutil.c.

References pg_hton64.

Referenced by sendFeedback().

 {
     uint64      n64 = pg_hton64(i);
 
     memcpy(buf, &n64, sizeof(n64));
 }

TimestampTz feGetCurrentTimestamp ( void )

Definition at line 520 of file streamutil.c.

References gettimeofday(), POSTGRES_EPOCH_JDATE, SECS_PER_DAY, UNIX_EPOCH_JDATE, and USECS_PER_SEC.

Referenced by flushAndSendFeedback(), HandleCopyStream(), ProcessKeepaliveMsg(), and StreamLogicalLog().

 {
     TimestampTz result;
     struct timeval tp;
 
     gettimeofday(&tp, NULL);
 
     result = (TimestampTz) tp.tv_sec -
         ((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY);
     result = (result * USECS_PER_SEC) + tp.tv_usec;
 
     return result;
 }

void feTimestampDifference	(	TimestampTz	start_time,
		TimestampTz	stop_time,
		long *	secs,
		int *	microsecs
	)

Definition at line 539 of file streamutil.c.

References start_time, and USECS_PER_SEC.

Referenced by CalculateCopyStreamSleeptime(), and StreamLogicalLog().

 {
     TimestampTz diff = stop_time - start_time;
 
     if (diff <= 0)
     {
         *secs = 0;
         *microsecs = 0;
     }
     else
     {
         *secs = (long) (diff / USECS_PER_SEC);
         *microsecs = (int) (diff % USECS_PER_SEC);
     }
 }

bool feTimestampDifferenceExceeds	(	TimestampTz	start_time,
		TimestampTz	stop_time,
		int	msec
	)

Definition at line 561 of file streamutil.c.

References start_time.

Referenced by HandleCopyStream(), and StreamLogicalLog().

 {
     TimestampTz diff = stop_time - start_time;
 
     return (diff >= msec * INT64CONST(1000));
 }

PGconn* GetConnection ( void )

Definition at line 54 of file streamutil.c.

References _, Assert, CONNECTION_BAD, CONNECTION_OK, connection_string, dbgetpassword, dbhost, dbname, dbport, dbuser, free, have_password, i, _PQconninfoOption::keyword, password, pg_malloc0(), PQconnectdbParams(), PQconnectionNeedsPassword(), PQconninfoFree(), PQconninfoParse(), PQerrorMessage(), PQfinish(), PQparameterStatus(), PQstatus(), progname, simple_prompt(), _PQconninfoOption::val, and values.

 {
     PGconn     *tmpconn;
     int         argcount = 7;   /* dbname, replication, fallback_app_name,
                                  * host, user, port, password */
     int         i;
     const char **keywords;
     const char **values;
     const char *tmpparam;
     bool        need_password;
     PQconninfoOption *conn_opts = NULL;
     PQconninfoOption *conn_opt;
     char       *err_msg = NULL;
 
     /* pg_recvlogical uses dbname only; others use connection_string only. */
     Assert(dbname == NULL || connection_string == NULL);
 
     /*
      * Merge the connection info inputs given in form of connection string,
      * options and default values (dbname=replication, replication=true, etc.)
      * Explicitly discard any dbname value in the connection string;
      * otherwise, PQconnectdbParams() would interpret that value as being
      * itself a connection string.
      */
     i = 0;
     if (connection_string)
     {
         conn_opts = PQconninfoParse(connection_string, &err_msg);
         if (conn_opts == NULL)
         {
             fprintf(stderr, "%s: %s", progname, err_msg);
             exit(1);
         }
 
         for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
         {
             if (conn_opt->val != NULL && conn_opt->val[0] != '\0' &&
                 strcmp(conn_opt->keyword, "dbname") != 0)
                 argcount++;
         }
 
         keywords = pg_malloc0((argcount + 1) * sizeof(*keywords));
         values = pg_malloc0((argcount + 1) * sizeof(*values));
 
         for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
         {
             if (conn_opt->val != NULL && conn_opt->val[0] != '\0' &&
                 strcmp(conn_opt->keyword, "dbname") != 0)
             {
                 keywords[i] = conn_opt->keyword;
                 values[i] = conn_opt->val;
                 i++;
             }
         }
     }
     else
     {
         keywords = pg_malloc0((argcount + 1) * sizeof(*keywords));
         values = pg_malloc0((argcount + 1) * sizeof(*values));
     }
 
     keywords[i] = "dbname";
     values[i] = dbname == NULL ? "replication" : dbname;
     i++;
     keywords[i] = "replication";
     values[i] = dbname == NULL ? "true" : "database";
     i++;
     keywords[i] = "fallback_application_name";
     values[i] = progname;
     i++;
 
     if (dbhost)
     {
         keywords[i] = "host";
         values[i] = dbhost;
         i++;
     }
     if (dbuser)
     {
         keywords[i] = "user";
         values[i] = dbuser;
         i++;
     }
     if (dbport)
     {
         keywords[i] = "port";
         values[i] = dbport;
         i++;
     }
 
     /* If -W was given, force prompt for password, but only the first time */
     need_password = (dbgetpassword == 1 && !have_password);
 
     do
     {
         /* Get a new password if appropriate */
         if (need_password)
         {
             simple_prompt("Password: ", password, sizeof(password), false);
             have_password = true;
             need_password = false;
         }
 
         /* Use (or reuse, on a subsequent connection) password if we have it */
         if (have_password)
         {
             keywords[i] = "password";
             values[i] = password;
         }
         else
         {
             keywords[i] = NULL;
             values[i] = NULL;
         }
 
         tmpconn = PQconnectdbParams(keywords, values, true);
 
         /*
          * If there is too little memory even to allocate the PGconn object
          * and PQconnectdbParams returns NULL, we call exit(1) directly.
          */
         if (!tmpconn)
         {
             fprintf(stderr, _("%s: could not connect to server\n"),
                     progname);
             exit(1);
         }
 
         /* If we need a password and -w wasn't given, loop back and get one */
         if (PQstatus(tmpconn) == CONNECTION_BAD &&
             PQconnectionNeedsPassword(tmpconn) &&
             dbgetpassword != -1)
         {
             PQfinish(tmpconn);
             need_password = true;
         }
     }
     while (need_password);
 
     if (PQstatus(tmpconn) != CONNECTION_OK)
     {
         fprintf(stderr, _("%s: could not connect to server: %s"),
                 progname, PQerrorMessage(tmpconn));
         PQfinish(tmpconn);
         free(values);
         free(keywords);
         if (conn_opts)
             PQconninfoFree(conn_opts);
         return NULL;
     }
 
     /* Connection ok! */
     free(values);
     free(keywords);
     if (conn_opts)
         PQconninfoFree(conn_opts);
 
     /*
      * Ensure we have the same value of integer_datetimes (now always "on") as
      * the server we are connecting to.
      */
     tmpparam = PQparameterStatus(tmpconn, "integer_datetimes");
     if (!tmpparam)
     {
         fprintf(stderr,
                 _("%s: could not determine server setting for integer_datetimes\n"),
                 progname);
         PQfinish(tmpconn);
         exit(1);
     }
 
     if (strcmp(tmpparam, "on") != 0)
     {
         fprintf(stderr,
                 _("%s: integer_datetimes compile flag does not match server\n"),
                 progname);
         PQfinish(tmpconn);
         exit(1);
     }
 
     return tmpconn;
 }

bool RetrieveWalSegSize ( PGconn * conn )

Definition at line 242 of file streamutil.c.

References _, Assert, DEFAULT_XLOG_SEG_SIZE, IsValidWalSegSize, MINIMUM_VERSION_FOR_SHOW_CMD, PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQexec(), PQgetvalue(), PQnfields(), PQntuples(), PQresultStatus(), PQserverVersion(), progname, and WalSegSz.

Referenced by main().

 {
     PGresult   *res;
     char        xlog_unit[3];
     int         xlog_val,
                 multiplier = 1;
 
     /* check connection existence */
     Assert(conn != NULL);
 
     /* for previous versions set the default xlog seg size */
     if (PQserverVersion(conn) < MINIMUM_VERSION_FOR_SHOW_CMD)
     {
         WalSegSz = DEFAULT_XLOG_SEG_SIZE;
         return true;
     }
 
     res = PQexec(conn, "SHOW wal_segment_size");
     if (PQresultStatus(res) != PGRES_TUPLES_OK)
     {
         fprintf(stderr, _("%s: could not send replication command \"%s\": %s\n"),
                 progname, "SHOW wal_segment_size", PQerrorMessage(conn));
 
         PQclear(res);
         return false;
     }
     if (PQntuples(res) != 1 || PQnfields(res) < 1)
     {
         fprintf(stderr,
                 _("%s: could not fetch WAL segment size: got %d rows and %d fields, expected %d rows and %d or more fields\n"),
                 progname, PQntuples(res), PQnfields(res), 1, 1);
 
         PQclear(res);
         return false;
     }
 
     /* fetch xlog value and unit from the result */
     if (sscanf(PQgetvalue(res, 0, 0), "%d%s", &xlog_val, xlog_unit) != 2)
     {
         fprintf(stderr, _("%s: WAL segment size could not be parsed\n"),
                 progname);
         return false;
     }
 
     /* set the multiplier based on unit to convert xlog_val to bytes */
     if (strcmp(xlog_unit, "MB") == 0)
         multiplier = 1024 * 1024;
     else if (strcmp(xlog_unit, "GB") == 0)
         multiplier = 1024 * 1024 * 1024;
 
     /* convert and set WalSegSz */
     WalSegSz = xlog_val * multiplier;
 
     if (!IsValidWalSegSize(WalSegSz))
     {
         fprintf(stderr,
                 _("%s: WAL segment size must be a power of two between 1MB and 1GB, but the remote server reported a value of %d bytes\n"),
                 progname, WalSegSz);
         return false;
     }
 
     PQclear(res);
     return true;
 }

bool RunIdentifySystem	(	PGconn *	conn,
		char **	sysid,
		TimeLineID *	starttli,
		XLogRecPtr *	startpos,
		char **	db_name
	)

Definition at line 316 of file streamutil.c.

References _, Assert, pg_strdup(), PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQexec(), PQgetisnull(), PQgetvalue(), PQnfields(), PQntuples(), PQresultStatus(), PQserverVersion(), and progname.

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

 {
     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));
 
         PQclear(res);
         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);
 
         PQclear(res);
         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 write-ahead log location \"%s\"\n"),
                     progname, PQgetvalue(res, 0, 2));
 
             PQclear(res);
             return false;
         }
         *startpos = ((uint64) hi) << 32 | lo;
     }
 
     /* Get database name, only available in 9.4 and newer versions */
     if (db_name != NULL)
     {
         *db_name = NULL;
         if (PQserverVersion(conn) >= 90400)
         {
             if (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);
 
                 PQclear(res);
                 return false;
             }
             if (!PQgetisnull(res, 0, 3))
                 *db_name = pg_strdup(PQgetvalue(res, 0, 3));
         }
     }
 
     PQclear(res);
     return true;
 }