granicus.if.org Git - postgresql/blob - src/backend/replication/libpqwalreceiver/libpqwalreceiver.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * libpqwalreceiver.c
   4  *
   5  * This file contains the libpq-specific parts of walreceiver. It's
   6  * loaded as a dynamic module to avoid linking the main server binary with
   7  * libpq.
   8  *
   9  * Portions Copyright (c) 2010-2014, PostgreSQL Global Development Group
  10  *
  11  *
  12  * IDENTIFICATION
  13  *        src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
  14  *
  15  *-------------------------------------------------------------------------
  16  */
  17 #include "postgres.h"
  18
  19 #include <unistd.h>
  20 #include <sys/time.h>
  21
  22 #include "libpq-fe.h"
  23 #include "access/xlog.h"
  24 #include "miscadmin.h"
  25 #include "replication/walreceiver.h"
  26 #include "utils/builtins.h"
  27
  28 #ifdef HAVE_POLL_H
  29 #include <poll.h>
  30 #endif
  31 #ifdef HAVE_SYS_POLL_H
  32 #include <sys/poll.h>
  33 #endif
  34 #ifdef HAVE_SYS_SELECT_H
  35 #include <sys/select.h>
  36 #endif
  37
  38 PG_MODULE_MAGIC;
  39
  40 void            _PG_init(void);
  41
  42 /* Current connection to the primary, if any */
  43 static PGconn *streamConn = NULL;
  44
  45 /* Buffer for currently read records */
  46 static char *recvBuf = NULL;
  47
  48 /* Prototypes for interface functions */
  49 static void libpqrcv_connect(char *conninfo);
  50 static void libpqrcv_identify_system(TimeLineID *primary_tli);
  51 static void libpqrcv_readtimelinehistoryfile(TimeLineID tli, char **filename, char **content, int *len);
  52 static bool libpqrcv_startstreaming(TimeLineID tli, XLogRecPtr startpoint,
  53                                                                         char *slotname);
  54 static void libpqrcv_endstreaming(TimeLineID *next_tli);
  55 static int      libpqrcv_receive(int timeout, char **buffer);
  56 static void libpqrcv_send(const char *buffer, int nbytes);
  57 static void libpqrcv_disconnect(void);
  58
  59 /* Prototypes for private functions */
  60 static bool libpq_select(int timeout_ms);
  61 static PGresult *libpqrcv_PQexec(const char *query);
  62
  63 /*
  64  * Module load callback
  65  */
  66 void
  67 _PG_init(void)
  68 {
  69         /* Tell walreceiver how to reach us */
  70         if (walrcv_connect != NULL || walrcv_identify_system != NULL ||
  71                 walrcv_readtimelinehistoryfile != NULL ||
  72                 walrcv_startstreaming != NULL || walrcv_endstreaming != NULL ||
  73                 walrcv_receive != NULL || walrcv_send != NULL ||
  74                 walrcv_disconnect != NULL)
  75                 elog(ERROR, "libpqwalreceiver already loaded");
  76         walrcv_connect = libpqrcv_connect;
  77         walrcv_identify_system = libpqrcv_identify_system;
  78         walrcv_readtimelinehistoryfile = libpqrcv_readtimelinehistoryfile;
  79         walrcv_startstreaming = libpqrcv_startstreaming;
  80         walrcv_endstreaming = libpqrcv_endstreaming;
  81         walrcv_receive = libpqrcv_receive;
  82         walrcv_send = libpqrcv_send;
  83         walrcv_disconnect = libpqrcv_disconnect;
  84 }
  85
  86 /*
  87  * Establish the connection to the primary server for XLOG streaming
  88  */
  89 static void
  90 libpqrcv_connect(char *conninfo)
  91 {
  92         char            conninfo_repl[MAXCONNINFO + 75];
  93
  94         /*
  95          * Connect using deliberately undocumented parameter: replication. The
  96          * database name is ignored by the server in replication mode, but specify
  97          * "replication" for .pgpass lookup.
  98          */
  99         snprintf(conninfo_repl, sizeof(conninfo_repl),
 100                          "%s dbname=replication replication=true fallback_application_name=walreceiver",
 101                          conninfo);
 102
 103         streamConn = PQconnectdb(conninfo_repl);
 104         if (PQstatus(streamConn) != CONNECTION_OK)
 105                 ereport(ERROR,
 106                                 (errmsg("could not connect to the primary server: %s",
 107                                                 PQerrorMessage(streamConn))));
 108 }
 109
 110 /*
 111  * Check that primary's system identifier matches ours, and fetch the current
 112  * timeline ID of the primary.
 113  */
 114 static void
 115 libpqrcv_identify_system(TimeLineID *primary_tli)
 116 {
 117         PGresult   *res;
 118         char       *primary_sysid;
 119         char            standby_sysid[32];
 120
 121         /*
 122          * Get the system identifier and timeline ID as a DataRow message from the
 123          * primary server.
 124          */
 125         res = libpqrcv_PQexec("IDENTIFY_SYSTEM");
 126         if (PQresultStatus(res) != PGRES_TUPLES_OK)
 127         {
 128                 PQclear(res);
 129                 ereport(ERROR,
 130                                 (errmsg("could not receive database system identifier and timeline ID from "
 131                                                 "the primary server: %s",
 132                                                 PQerrorMessage(streamConn))));
 133         }
 134         if (PQnfields(res) != 3 || PQntuples(res) != 1)
 135         {
 136                 int                     ntuples = PQntuples(res);
 137                 int                     nfields = PQnfields(res);
 138
 139                 PQclear(res);
 140                 ereport(ERROR,
 141                                 (errmsg("invalid response from primary server"),
 142                                  errdetail("Expected 1 tuple with 3 fields, got %d tuples with %d fields.",
 143                                                    ntuples, nfields)));
 144         }
 145         primary_sysid = PQgetvalue(res, 0, 0);
 146         *primary_tli = pg_atoi(PQgetvalue(res, 0, 1), 4, 0);
 147
 148         /*
 149          * Confirm that the system identifier of the primary is the same as ours.
 150          */
 151         snprintf(standby_sysid, sizeof(standby_sysid), UINT64_FORMAT,
 152                          GetSystemIdentifier());
 153         if (strcmp(primary_sysid, standby_sysid) != 0)
 154         {
 155                 PQclear(res);
 156                 ereport(ERROR,
 157                                 (errmsg("database system identifier differs between the primary and standby"),
 158                                  errdetail("The primary's identifier is %s, the standby's identifier is %s.",
 159                                                    primary_sysid, standby_sysid)));
 160         }
 161         PQclear(res);
 162 }
 163
 164 /*
 165  * Start streaming WAL data from given startpoint and timeline.
 166  *
 167  * Returns true if we switched successfully to copy-both mode. False
 168  * means the server received the command and executed it successfully, but
 169  * didn't switch to copy-mode.  That means that there was no WAL on the
 170  * requested timeline and starting point, because the server switched to
 171  * another timeline at or before the requested starting point. On failure,
 172  * throws an ERROR.
 173  */
 174 static bool
 175 libpqrcv_startstreaming(TimeLineID tli, XLogRecPtr startpoint, char *slotname)
 176 {
 177         char            cmd[64];
 178         PGresult   *res;
 179
 180         /* Start streaming from the point requested by startup process */
 181         if (slotname != NULL)
 182                 snprintf(cmd, sizeof(cmd),
 183                                  "START_REPLICATION SLOT \"%s\" %X/%X TIMELINE %u", slotname,
 184                                  (uint32) (startpoint >> 32), (uint32) startpoint, tli);
 185         else
 186                 snprintf(cmd, sizeof(cmd),
 187                                  "START_REPLICATION %X/%X TIMELINE %u",
 188                                  (uint32) (startpoint >> 32), (uint32) startpoint, tli);
 189         res = libpqrcv_PQexec(cmd);
 190
 191         if (PQresultStatus(res) == PGRES_COMMAND_OK)
 192         {
 193                 PQclear(res);
 194                 return false;
 195         }
 196         else if (PQresultStatus(res) != PGRES_COPY_BOTH)
 197         {
 198                 PQclear(res);
 199                 ereport(ERROR,
 200                                 (errmsg("could not start WAL streaming: %s",
 201                                                 PQerrorMessage(streamConn))));
 202         }
 203         PQclear(res);
 204         return true;
 205 }
 206
 207 /*
 208  * Stop streaming WAL data. Returns the next timeline's ID in *next_tli, as
 209  * reported by the server, or 0 if it did not report it.
 210  */
 211 static void
 212 libpqrcv_endstreaming(TimeLineID *next_tli)
 213 {
 214         PGresult   *res;
 215
 216         if (PQputCopyEnd(streamConn, NULL) <= 0 || PQflush(streamConn))
 217                 ereport(ERROR,
 218                         (errmsg("could not send end-of-streaming message to primary: %s",
 219                                         PQerrorMessage(streamConn))));
 220
 221         /*
 222          * After COPY is finished, we should receive a result set indicating the
 223          * next timeline's ID, or just CommandComplete if the server was shut
 224          * down.
 225          *
 226          * If we had not yet received CopyDone from the backend, PGRES_COPY_IN
 227          * would also be possible. However, at the moment this function is only
 228          * called after receiving CopyDone from the backend - the walreceiver
 229          * never terminates replication on its own initiative.
 230          */
 231         res = PQgetResult(streamConn);
 232         if (PQresultStatus(res) == PGRES_TUPLES_OK)
 233         {
 234                 /*
 235                  * Read the next timeline's ID. The server also sends the timeline's
 236                  * starting point, but it is ignored.
 237                  */
 238                 if (PQnfields(res) < 2 || PQntuples(res) != 1)
 239                         ereport(ERROR,
 240                                         (errmsg("unexpected result set after end-of-streaming")));
 241                 *next_tli = pg_atoi(PQgetvalue(res, 0, 0), sizeof(uint32), 0);
 242                 PQclear(res);
 243
 244                 /* the result set should be followed by CommandComplete */
 245                 res = PQgetResult(streamConn);
 246         }
 247         else
 248                 *next_tli = 0;
 249
 250         if (PQresultStatus(res) != PGRES_COMMAND_OK)
 251                 ereport(ERROR,
 252                                 (errmsg("error reading result of streaming command: %s",
 253                                                 PQerrorMessage(streamConn))));
 254
 255         /* Verify that there are no more results */
 256         res = PQgetResult(streamConn);
 257         if (res != NULL)
 258                 ereport(ERROR,
 259                                 (errmsg("unexpected result after CommandComplete: %s",
 260                                                 PQerrorMessage(streamConn))));
 261 }
 262
 263 /*
 264  * Fetch the timeline history file for 'tli' from primary.
 265  */
 266 static void
 267 libpqrcv_readtimelinehistoryfile(TimeLineID tli,
 268                                                                  char **filename, char **content, int *len)
 269 {
 270         PGresult   *res;
 271         char            cmd[64];
 272
 273         /*
 274          * Request the primary to send over the history file for given timeline.
 275          */
 276         snprintf(cmd, sizeof(cmd), "TIMELINE_HISTORY %u", tli);
 277         res = libpqrcv_PQexec(cmd);
 278         if (PQresultStatus(res) != PGRES_TUPLES_OK)
 279         {
 280                 PQclear(res);
 281                 ereport(ERROR,
 282                                 (errmsg("could not receive timeline history file from "
 283                                                 "the primary server: %s",
 284                                                 PQerrorMessage(streamConn))));
 285         }
 286         if (PQnfields(res) != 2 || PQntuples(res) != 1)
 287         {
 288                 int                     ntuples = PQntuples(res);
 289                 int                     nfields = PQnfields(res);
 290
 291                 PQclear(res);
 292                 ereport(ERROR,
 293                                 (errmsg("invalid response from primary server"),
 294                                  errdetail("Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
 295                                                    ntuples, nfields)));
 296         }
 297         *filename = pstrdup(PQgetvalue(res, 0, 0));
 298
 299         *len = PQgetlength(res, 0, 1);
 300         *content = palloc(*len);
 301         memcpy(*content, PQgetvalue(res, 0, 1), *len);
 302         PQclear(res);
 303 }
 304
 305 /*
 306  * Wait until we can read WAL stream, or timeout.
 307  *
 308  * Returns true if data has become available for reading, false if timed out
 309  * or interrupted by signal.
 310  *
 311  * This is based on pqSocketCheck.
 312  */
 313 static bool
 314 libpq_select(int timeout_ms)
 315 {
 316         int                     ret;
 317
 318         Assert(streamConn != NULL);
 319         if (PQsocket(streamConn) < 0)
 320                 ereport(ERROR,
 321                                 (errcode_for_socket_access(),
 322                                  errmsg("socket not open")));
 323
 324         /* We use poll(2) if available, otherwise select(2) */
 325         {
 326 #ifdef HAVE_POLL
 327                 struct pollfd input_fd;
 328
 329                 input_fd.fd = PQsocket(streamConn);
 330                 input_fd.events = POLLIN | POLLERR;
 331                 input_fd.revents = 0;
 332
 333                 ret = poll(&input_fd, 1, timeout_ms);
 334 #else                                                   /* !HAVE_POLL */
 335
 336                 fd_set          input_mask;
 337                 struct timeval timeout;
 338                 struct timeval *ptr_timeout;
 339
 340                 FD_ZERO(&input_mask);
 341                 FD_SET(PQsocket(streamConn), &input_mask);
 342
 343                 if (timeout_ms < 0)
 344                         ptr_timeout = NULL;
 345                 else
 346                 {
 347                         timeout.tv_sec = timeout_ms / 1000;
 348                         timeout.tv_usec = (timeout_ms % 1000) * 1000;
 349                         ptr_timeout = &timeout;
 350                 }
 351
 352                 ret = select(PQsocket(streamConn) + 1, &input_mask,
 353                                          NULL, NULL, ptr_timeout);
 354 #endif   /* HAVE_POLL */
 355         }
 356
 357         if (ret == 0 || (ret < 0 && errno == EINTR))
 358                 return false;
 359         if (ret < 0)
 360                 ereport(ERROR,
 361                                 (errcode_for_socket_access(),
 362                                  errmsg("select() failed: %m")));
 363         return true;
 364 }
 365
 366 /*
 367  * Send a query and wait for the results by using the asynchronous libpq
 368  * functions and the backend version of select().
 369  *
 370  * We must not use the regular blocking libpq functions like PQexec()
 371  * since they are uninterruptible by signals on some platforms, such as
 372  * Windows.
 373  *
 374  * We must also not use vanilla select() here since it cannot handle the
 375  * signal emulation layer on Windows.
 376  *
 377  * The function is modeled on PQexec() in libpq, but only implements
 378  * those parts that are in use in the walreceiver.
 379  *
 380  * Queries are always executed on the connection in streamConn.
 381  */
 382 static PGresult *
 383 libpqrcv_PQexec(const char *query)
 384 {
 385         PGresult   *result = NULL;
 386         PGresult   *lastResult = NULL;
 387
 388         /*
 389          * PQexec() silently discards any prior query results on the connection.
 390          * This is not required for walreceiver since it's expected that walsender
 391          * won't generate any such junk results.
 392          */
 393
 394         /*
 395          * Submit a query. Since we don't use non-blocking mode, this also can
 396          * block. But its risk is relatively small, so we ignore that for now.
 397          */
 398         if (!PQsendQuery(streamConn, query))
 399                 return NULL;
 400
 401         for (;;)
 402         {
 403                 /*
 404                  * Receive data until PQgetResult is ready to get the result without
 405                  * blocking.
 406                  */
 407                 while (PQisBusy(streamConn))
 408                 {
 409                         /*
 410                          * We don't need to break down the sleep into smaller increments,
 411                          * and check for interrupts after each nap, since we can just
 412                          * elog(FATAL) within SIGTERM signal handler if the signal arrives
 413                          * in the middle of establishment of replication connection.
 414                          */
 415                         if (!libpq_select(-1))
 416                                 continue;               /* interrupted */
 417                         if (PQconsumeInput(streamConn) == 0)
 418                                 return NULL;    /* trouble */
 419                 }
 420
 421                 /*
 422                  * Emulate the PQexec()'s behavior of returning the last result when
 423                  * there are many. Since walsender will never generate multiple
 424                  * results, we skip the concatenation of error messages.
 425                  */
 426                 result = PQgetResult(streamConn);
 427                 if (result == NULL)
 428                         break;                          /* query is complete */
 429
 430                 PQclear(lastResult);
 431                 lastResult = result;
 432
 433                 if (PQresultStatus(lastResult) == PGRES_COPY_IN ||
 434                         PQresultStatus(lastResult) == PGRES_COPY_OUT ||
 435                         PQresultStatus(lastResult) == PGRES_COPY_BOTH ||
 436                         PQstatus(streamConn) == CONNECTION_BAD)
 437                         break;
 438         }
 439
 440         return lastResult;
 441 }
 442
 443 /*
 444  * Disconnect connection to primary, if any.
 445  */
 446 static void
 447 libpqrcv_disconnect(void)
 448 {
 449         PQfinish(streamConn);
 450         streamConn = NULL;
 451 }
 452
 453 /*
 454  * Receive a message available from XLOG stream, blocking for
 455  * maximum of 'timeout' ms.
 456  *
 457  * Returns:
 458  *
 459  *       If data was received, returns the length of the data. *buffer is set to
 460  *       point to a buffer holding the received message. The buffer is only valid
 461  *       until the next libpqrcv_* call.
 462  *
 463  *       0 if no data was available within timeout, or wait was interrupted
 464  *       by signal.
 465  *
 466  *       -1 if the server ended the COPY.
 467  *
 468  * ereports on error.
 469  */
 470 static int
 471 libpqrcv_receive(int timeout, char **buffer)
 472 {
 473         int                     rawlen;
 474
 475         if (recvBuf != NULL)
 476                 PQfreemem(recvBuf);
 477         recvBuf = NULL;
 478
 479         /* Try to receive a CopyData message */
 480         rawlen = PQgetCopyData(streamConn, &recvBuf, 1);
 481         if (rawlen == 0)
 482         {
 483                 /*
 484                  * No data available yet. If the caller requested to block, wait for
 485                  * more data to arrive.
 486                  */
 487                 if (timeout > 0)
 488                 {
 489                         if (!libpq_select(timeout))
 490                                 return 0;
 491                 }
 492
 493                 if (PQconsumeInput(streamConn) == 0)
 494                         ereport(ERROR,
 495                                         (errmsg("could not receive data from WAL stream: %s",
 496                                                         PQerrorMessage(streamConn))));
 497
 498                 /* Now that we've consumed some input, try again */
 499                 rawlen = PQgetCopyData(streamConn, &recvBuf, 1);
 500                 if (rawlen == 0)
 501                         return 0;
 502         }
 503         if (rawlen == -1)                       /* end-of-streaming or error */
 504         {
 505                 PGresult   *res;
 506
 507                 res = PQgetResult(streamConn);
 508                 if (PQresultStatus(res) == PGRES_COMMAND_OK ||
 509                         PQresultStatus(res) == PGRES_COPY_IN)
 510                 {
 511                         PQclear(res);
 512                         return -1;
 513                 }
 514                 else
 515                 {
 516                         PQclear(res);
 517                         ereport(ERROR,
 518                                         (errmsg("could not receive data from WAL stream: %s",
 519                                                         PQerrorMessage(streamConn))));
 520                 }
 521         }
 522         if (rawlen < -1)
 523                 ereport(ERROR,
 524                                 (errmsg("could not receive data from WAL stream: %s",
 525                                                 PQerrorMessage(streamConn))));
 526
 527         /* Return received messages to caller */
 528         *buffer = recvBuf;
 529         return rawlen;
 530 }
 531
 532 /*
 533  * Send a message to XLOG stream.
 534  *
 535  * ereports on error.
 536  */
 537 static void
 538 libpqrcv_send(const char *buffer, int nbytes)
 539 {
 540         if (PQputCopyData(streamConn, buffer, nbytes) <= 0 ||
 541                 PQflush(streamConn))
 542                 ereport(ERROR,
 543                                 (errmsg("could not send data to WAL stream: %s",
 544                                                 PQerrorMessage(streamConn))));
 545 }