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

   1 /*-------------------------------------------------------------------------
   2  *
   3  * walsender.c
   4  *
   5  * The WAL sender process (walsender) is new as of Postgres 9.0. It takes
   6  * care of sending XLOG from the primary server to a single recipient.
   7  * (Note that there can be more than one walsender process concurrently.)
   8  * It is started by the postmaster when the walreceiver of a standby server
   9  * connects to the primary server and requests XLOG streaming replication.
  10  * It attempts to keep reading XLOG records from the disk and sending them
  11  * to the standby server, as long as the connection is alive (i.e., like
  12  * any backend, there is a one-to-one relationship between a connection
  13  * and a walsender process).
  14  *
  15  * Normal termination is by SIGTERM, which instructs the walsender to
  16  * close the connection and exit(0) at next convenient moment. Emergency
  17  * termination is by SIGQUIT; like any backend, the walsender will simply
  18  * abort and exit on SIGQUIT. A close of the connection and a FATAL error
  19  * are treated as not a crash but approximately normal termination;
  20  * the walsender will exit quickly without sending any more XLOG records.
  21  *
  22  * If the server is shut down, postmaster sends us SIGUSR2 after all
  23  * regular backends have exited and the shutdown checkpoint has been written.
  24  * This instruct walsender to send any outstanding WAL, including the
  25  * shutdown checkpoint record, and then exit.
  26  *
  27  *
  28  * Portions Copyright (c) 2010-2012, PostgreSQL Global Development Group
  29  *
  30  * IDENTIFICATION
  31  *        src/backend/replication/walsender.c
  32  *
  33  *-------------------------------------------------------------------------
  34  */
  35 #include "postgres.h"
  36
  37 #include <signal.h>
  38 #include <unistd.h>
  39
  40 #include "access/transam.h"
  41 #include "access/xlog_internal.h"
  42 #include "catalog/pg_type.h"
  43 #include "funcapi.h"
  44 #include "libpq/libpq.h"
  45 #include "libpq/pqformat.h"
  46 #include "libpq/pqsignal.h"
  47 #include "miscadmin.h"
  48 #include "nodes/replnodes.h"
  49 #include "replication/basebackup.h"
  50 #include "replication/syncrep.h"
  51 #include "replication/walprotocol.h"
  52 #include "replication/walreceiver.h"
  53 #include "replication/walsender.h"
  54 #include "replication/walsender_private.h"
  55 #include "storage/fd.h"
  56 #include "storage/ipc.h"
  57 #include "storage/pmsignal.h"
  58 #include "storage/proc.h"
  59 #include "storage/procarray.h"
  60 #include "tcop/tcopprot.h"
  61 #include "utils/builtins.h"
  62 #include "utils/guc.h"
  63 #include "utils/memutils.h"
  64 #include "utils/ps_status.h"
  65 #include "utils/resowner.h"
  66 #include "utils/timestamp.h"
  67
  68
  69 /* Array of WalSnds in shared memory */
  70 WalSndCtlData *WalSndCtl = NULL;
  71
  72 /* My slot in the shared memory array */
  73 WalSnd     *MyWalSnd = NULL;
  74
  75 /* Global state */
  76 bool            am_walsender = false;           /* Am I a walsender process ? */
  77 bool            am_cascading_walsender = false;         /* Am I cascading WAL to
  78                                                                                                  * another standby ? */
  79
  80 /* User-settable parameters for walsender */
  81 int                     max_wal_senders = 0;    /* the maximum number of concurrent walsenders */
  82 int                     replication_timeout = 60 * 1000;        /* maximum time to send one
  83                                                                                                  * WAL data message */
  84 /*
  85  * State for WalSndWakeupRequest
  86  */
  87 bool wake_wal_senders = false;
  88
  89 /*
  90  * These variables are used similarly to openLogFile/Id/Seg/Off,
  91  * but for walsender to read the XLOG.
  92  */
  93 static int      sendFile = -1;
  94 static XLogSegNo sendSegNo = 0;
  95 static uint32 sendOff = 0;
  96
  97 /*
  98  * How far have we sent WAL already? This is also advertised in
  99  * MyWalSnd->sentPtr.  (Actually, this is the next WAL location to send.)
 100  */
 101 static XLogRecPtr sentPtr = 0;
 102
 103 /*
 104  * Buffer for processing reply messages.
 105  */
 106 static StringInfoData reply_message;
 107
 108 /*
 109  * Timestamp of the last receipt of the reply from the standby.
 110  */
 111 static TimestampTz last_reply_timestamp;
 112
 113 /* Flags set by signal handlers for later service in main loop */
 114 static volatile sig_atomic_t got_SIGHUP = false;
 115 volatile sig_atomic_t walsender_shutdown_requested = false;
 116 volatile sig_atomic_t walsender_ready_to_stop = false;
 117
 118 /* Signal handlers */
 119 static void WalSndSigHupHandler(SIGNAL_ARGS);
 120 static void WalSndShutdownHandler(SIGNAL_ARGS);
 121 static void WalSndQuickDieHandler(SIGNAL_ARGS);
 122 static void WalSndXLogSendHandler(SIGNAL_ARGS);
 123 static void WalSndLastCycleHandler(SIGNAL_ARGS);
 124
 125 /* Prototypes for private functions */
 126 static bool HandleReplicationCommand(const char *cmd_string);
 127 static void WalSndLoop(void) __attribute__((noreturn));
 128 static void InitWalSnd(void);
 129 static void WalSndHandshake(void);
 130 static void WalSndKill(int code, Datum arg);
 131 static void XLogSend(char *msgbuf, bool *caughtup);
 132 static void IdentifySystem(void);
 133 static void StartReplication(StartReplicationCmd *cmd);
 134 static void ProcessStandbyMessage(void);
 135 static void ProcessStandbyReplyMessage(void);
 136 static void ProcessStandbyHSFeedbackMessage(void);
 137 static void ProcessRepliesIfAny(void);
 138 static void WalSndKeepalive(char *msgbuf);
 139
 140
 141 /* Main entry point for walsender process */
 142 void
 143 WalSenderMain(void)
 144 {
 145         MemoryContext walsnd_context;
 146
 147         am_cascading_walsender = RecoveryInProgress();
 148
 149         /* Create a per-walsender data structure in shared memory */
 150         InitWalSnd();
 151
 152         /*
 153          * Create a memory context that we will do all our work in.  We do this so
 154          * that we can reset the context during error recovery and thereby avoid
 155          * possible memory leaks.  Formerly this code just ran in
 156          * TopMemoryContext, but resetting that would be a really bad idea.
 157          *
 158          * XXX: we don't actually attempt error recovery in walsender, we just
 159          * close the connection and exit.
 160          */
 161         walsnd_context = AllocSetContextCreate(TopMemoryContext,
 162                                                                                    "Wal Sender",
 163                                                                                    ALLOCSET_DEFAULT_MINSIZE,
 164                                                                                    ALLOCSET_DEFAULT_INITSIZE,
 165                                                                                    ALLOCSET_DEFAULT_MAXSIZE);
 166         MemoryContextSwitchTo(walsnd_context);
 167
 168         /* Set up resource owner */
 169         CurrentResourceOwner = ResourceOwnerCreate(NULL, "walsender top-level resource owner");
 170
 171         /* Unblock signals (they were blocked when the postmaster forked us) */
 172         PG_SETMASK(&UnBlockSig);
 173
 174         /*
 175          * Use the recovery target timeline ID during recovery
 176          */
 177         if (am_cascading_walsender)
 178                 ThisTimeLineID = GetRecoveryTargetTLI();
 179
 180         /* Tell the standby that walsender is ready for receiving commands */
 181         ReadyForQuery(DestRemote);
 182
 183         /* Handle handshake messages before streaming */
 184         WalSndHandshake();
 185
 186         /* Initialize shared memory status */
 187         {
 188                 /* use volatile pointer to prevent code rearrangement */
 189                 volatile WalSnd *walsnd = MyWalSnd;
 190
 191                 SpinLockAcquire(&walsnd->mutex);
 192                 walsnd->sentPtr = sentPtr;
 193                 SpinLockRelease(&walsnd->mutex);
 194         }
 195
 196         SyncRepInitConfig();
 197
 198         /* Main loop of walsender */
 199         WalSndLoop();
 200 }
 201
 202 /*
 203  * Execute commands from walreceiver, until we enter streaming mode.
 204  */
 205 static void
 206 WalSndHandshake(void)
 207 {
 208         StringInfoData input_message;
 209         bool            replication_started = false;
 210
 211         initStringInfo(&input_message);
 212
 213         while (!replication_started)
 214         {
 215                 int                     firstchar;
 216
 217                 WalSndSetState(WALSNDSTATE_STARTUP);
 218                 set_ps_display("idle", false);
 219
 220                 /* Wait for a command to arrive */
 221                 firstchar = pq_getbyte();
 222
 223                 /*
 224                  * Emergency bailout if postmaster has died.  This is to avoid the
 225                  * necessity for manual cleanup of all postmaster children.
 226                  */
 227                 if (!PostmasterIsAlive())
 228                         exit(1);
 229
 230                 /*
 231                  * Check for any other interesting events that happened while we
 232                  * slept.
 233                  */
 234                 if (got_SIGHUP)
 235                 {
 236                         got_SIGHUP = false;
 237                         ProcessConfigFile(PGC_SIGHUP);
 238                 }
 239
 240                 if (firstchar != EOF)
 241                 {
 242                         /*
 243                          * Read the message contents. This is expected to be done without
 244                          * blocking because we've been able to get message type code.
 245                          */
 246                         if (pq_getmessage(&input_message, 0))
 247                                 firstchar = EOF;        /* suitable message already logged */
 248                 }
 249
 250                 /* Handle the very limited subset of commands expected in this phase */
 251                 switch (firstchar)
 252                 {
 253                         case 'Q':                       /* Query message */
 254                                 {
 255                                         const char *query_string;
 256
 257                                         query_string = pq_getmsgstring(&input_message);
 258                                         pq_getmsgend(&input_message);
 259
 260                                         if (HandleReplicationCommand(query_string))
 261                                                 replication_started = true;
 262                                 }
 263                                 break;
 264
 265                         case 'X':
 266                                 /* standby is closing the connection */
 267                                 proc_exit(0);
 268
 269                         case EOF:
 270                                 /* standby disconnected unexpectedly */
 271                                 ereport(COMMERROR,
 272                                                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
 273                                                  errmsg("unexpected EOF on standby connection")));
 274                                 proc_exit(0);
 275
 276                         default:
 277                                 ereport(FATAL,
 278                                                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
 279                                                  errmsg("invalid standby handshake message type %d", firstchar)));
 280                 }
 281         }
 282 }
 283
 284 /*
 285  * IDENTIFY_SYSTEM
 286  */
 287 static void
 288 IdentifySystem(void)
 289 {
 290         StringInfoData buf;
 291         char            sysid[32];
 292         char            tli[11];
 293         char            xpos[MAXFNAMELEN];
 294         XLogRecPtr      logptr;
 295
 296         /*
 297          * Reply with a result set with one row, three columns. First col is
 298          * system ID, second is timeline ID, and third is current xlog location.
 299          */
 300
 301         snprintf(sysid, sizeof(sysid), UINT64_FORMAT,
 302                          GetSystemIdentifier());
 303         snprintf(tli, sizeof(tli), "%u", ThisTimeLineID);
 304
 305         logptr = am_cascading_walsender ? GetStandbyFlushRecPtr() : GetInsertRecPtr();
 306
 307         snprintf(xpos, sizeof(xpos), "%X/%X", (uint32) (logptr >> 32), (uint32) logptr);
 308
 309         /* Send a RowDescription message */
 310         pq_beginmessage(&buf, 'T');
 311         pq_sendint(&buf, 3, 2);         /* 3 fields */
 312
 313         /* first field */
 314         pq_sendstring(&buf, "systemid");        /* col name */
 315         pq_sendint(&buf, 0, 4);         /* table oid */
 316         pq_sendint(&buf, 0, 2);         /* attnum */
 317         pq_sendint(&buf, TEXTOID, 4);           /* type oid */
 318         pq_sendint(&buf, -1, 2);        /* typlen */
 319         pq_sendint(&buf, 0, 4);         /* typmod */
 320         pq_sendint(&buf, 0, 2);         /* format code */
 321
 322         /* second field */
 323         pq_sendstring(&buf, "timeline");        /* col name */
 324         pq_sendint(&buf, 0, 4);         /* table oid */
 325         pq_sendint(&buf, 0, 2);         /* attnum */
 326         pq_sendint(&buf, INT4OID, 4);           /* type oid */
 327         pq_sendint(&buf, 4, 2);         /* typlen */
 328         pq_sendint(&buf, 0, 4);         /* typmod */
 329         pq_sendint(&buf, 0, 2);         /* format code */
 330
 331         /* third field */
 332         pq_sendstring(&buf, "xlogpos");
 333         pq_sendint(&buf, 0, 4);
 334         pq_sendint(&buf, 0, 2);
 335         pq_sendint(&buf, TEXTOID, 4);
 336         pq_sendint(&buf, -1, 2);
 337         pq_sendint(&buf, 0, 4);
 338         pq_sendint(&buf, 0, 2);
 339         pq_endmessage(&buf);
 340
 341         /* Send a DataRow message */
 342         pq_beginmessage(&buf, 'D');
 343         pq_sendint(&buf, 3, 2);         /* # of columns */
 344         pq_sendint(&buf, strlen(sysid), 4); /* col1 len */
 345         pq_sendbytes(&buf, (char *) &sysid, strlen(sysid));
 346         pq_sendint(&buf, strlen(tli), 4);       /* col2 len */
 347         pq_sendbytes(&buf, (char *) tli, strlen(tli));
 348         pq_sendint(&buf, strlen(xpos), 4);      /* col3 len */
 349         pq_sendbytes(&buf, (char *) xpos, strlen(xpos));
 350
 351         pq_endmessage(&buf);
 352
 353         /* Send CommandComplete and ReadyForQuery messages */
 354         EndCommand("SELECT", DestRemote);
 355         ReadyForQuery(DestRemote);
 356         /* ReadyForQuery did pq_flush for us */
 357 }
 358
 359 /*
 360  * START_REPLICATION
 361  */
 362 static void
 363 StartReplication(StartReplicationCmd *cmd)
 364 {
 365         StringInfoData buf;
 366
 367         /*
 368          * Let postmaster know that we're streaming. Once we've declared us as a
 369          * WAL sender process, postmaster will let us outlive the bgwriter and
 370          * kill us last in the shutdown sequence, so we get a chance to stream all
 371          * remaining WAL at shutdown, including the shutdown checkpoint. Note that
 372          * there's no going back, and we mustn't write any WAL records after this.
 373          */
 374         MarkPostmasterChildWalSender();
 375         SendPostmasterSignal(PMSIGNAL_ADVANCE_STATE_MACHINE);
 376
 377         /*
 378          * When promoting a cascading standby, postmaster sends SIGUSR2 to any
 379          * cascading walsenders to kill them. But there is a corner-case where
 380          * such walsender fails to receive SIGUSR2 and survives a standby
 381          * promotion unexpectedly. This happens when postmaster sends SIGUSR2
 382          * before the walsender marks itself as a WAL sender, because postmaster
 383          * sends SIGUSR2 to only the processes marked as a WAL sender.
 384          *
 385          * To avoid this corner-case, if recovery is NOT in progress even though
 386          * the walsender is cascading one, we do the same thing as SIGUSR2 signal
 387          * handler does, i.e., set walsender_ready_to_stop to true. Which causes
 388          * the walsender to end later.
 389          *
 390          * When terminating cascading walsenders, usually postmaster writes the
 391          * log message announcing the terminations. But there is a race condition
 392          * here. If there is no walsender except this process before reaching
 393          * here, postmaster thinks that there is no walsender and suppresses that
 394          * log message. To handle this case, we always emit that log message here.
 395          * This might cause duplicate log messages, but which is less likely to
 396          * happen, so it's not worth writing some code to suppress them.
 397          */
 398         if (am_cascading_walsender && !RecoveryInProgress())
 399         {
 400                 ereport(LOG,
 401                    (errmsg("terminating walsender process to force cascaded standby "
 402                                    "to update timeline and reconnect")));
 403                 walsender_ready_to_stop = true;
 404         }
 405
 406         /*
 407          * We assume here that we're logging enough information in the WAL for
 408          * log-shipping, since this is checked in PostmasterMain().
 409          *
 410          * NOTE: wal_level can only change at shutdown, so in most cases it is
 411          * difficult for there to be WAL data that we can still see that was
 412          * written at wal_level='minimal'.
 413          */
 414
 415         /*
 416          * When we first start replication the standby will be behind the primary.
 417          * For some applications, for example, synchronous replication, it is
 418          * important to have a clear state for this initial catchup mode, so we
 419          * can trigger actions when we change streaming state later. We may stay
 420          * in this state for a long time, which is exactly why we want to be able
 421          * to monitor whether or not we are still here.
 422          */
 423         WalSndSetState(WALSNDSTATE_CATCHUP);
 424
 425         /* Send a CopyBothResponse message, and start streaming */
 426         pq_beginmessage(&buf, 'W');
 427         pq_sendbyte(&buf, 0);
 428         pq_sendint(&buf, 0, 2);
 429         pq_endmessage(&buf);
 430         pq_flush();
 431
 432         /*
 433          * Initialize position to the received one, then the xlog records begin to
 434          * be shipped from that position
 435          */
 436         sentPtr = cmd->startpoint;
 437 }
 438
 439 /*
 440  * Execute an incoming replication command.
 441  */
 442 static bool
 443 HandleReplicationCommand(const char *cmd_string)
 444 {
 445         bool            replication_started = false;
 446         int                     parse_rc;
 447         Node       *cmd_node;
 448         MemoryContext cmd_context;
 449         MemoryContext old_context;
 450
 451         elog(DEBUG1, "received replication command: %s", cmd_string);
 452
 453         cmd_context = AllocSetContextCreate(CurrentMemoryContext,
 454                                                                                 "Replication command context",
 455                                                                                 ALLOCSET_DEFAULT_MINSIZE,
 456                                                                                 ALLOCSET_DEFAULT_INITSIZE,
 457                                                                                 ALLOCSET_DEFAULT_MAXSIZE);
 458         old_context = MemoryContextSwitchTo(cmd_context);
 459
 460         replication_scanner_init(cmd_string);
 461         parse_rc = replication_yyparse();
 462         if (parse_rc != 0)
 463                 ereport(ERROR,
 464                                 (errcode(ERRCODE_SYNTAX_ERROR),
 465                                  (errmsg_internal("replication command parser returned %d",
 466                                                                   parse_rc))));
 467
 468         cmd_node = replication_parse_result;
 469
 470         switch (cmd_node->type)
 471         {
 472                 case T_IdentifySystemCmd:
 473                         IdentifySystem();
 474                         break;
 475
 476                 case T_StartReplicationCmd:
 477                         StartReplication((StartReplicationCmd *) cmd_node);
 478
 479                         /* break out of the loop */
 480                         replication_started = true;
 481                         break;
 482
 483                 case T_BaseBackupCmd:
 484                         SendBaseBackup((BaseBackupCmd *) cmd_node);
 485
 486                         /* Send CommandComplete and ReadyForQuery messages */
 487                         EndCommand("SELECT", DestRemote);
 488                         ReadyForQuery(DestRemote);
 489                         /* ReadyForQuery did pq_flush for us */
 490                         break;
 491
 492                 default:
 493                         ereport(FATAL,
 494                                         (errcode(ERRCODE_PROTOCOL_VIOLATION),
 495                                          errmsg("invalid standby query string: %s", cmd_string)));
 496         }
 497
 498         /* done */
 499         MemoryContextSwitchTo(old_context);
 500         MemoryContextDelete(cmd_context);
 501
 502         return replication_started;
 503 }
 504
 505 /*
 506  * Check if the remote end has closed the connection.
 507  */
 508 static void
 509 ProcessRepliesIfAny(void)
 510 {
 511         unsigned char firstchar;
 512         int                     r;
 513         bool            received = false;
 514
 515         for (;;)
 516         {
 517                 r = pq_getbyte_if_available(&firstchar);
 518                 if (r < 0)
 519                 {
 520                         /* unexpected error or EOF */
 521                         ereport(COMMERROR,
 522                                         (errcode(ERRCODE_PROTOCOL_VIOLATION),
 523                                          errmsg("unexpected EOF on standby connection")));
 524                         proc_exit(0);
 525                 }
 526                 if (r == 0)
 527                 {
 528                         /* no data available without blocking */
 529                         break;
 530                 }
 531
 532                 /* Handle the very limited subset of commands expected in this phase */
 533                 switch (firstchar)
 534                 {
 535                                 /*
 536                                  * 'd' means a standby reply wrapped in a CopyData packet.
 537                                  */
 538                         case 'd':
 539                                 ProcessStandbyMessage();
 540                                 received = true;
 541                                 break;
 542
 543                                 /*
 544                                  * 'X' means that the standby is closing down the socket.
 545                                  */
 546                         case 'X':
 547                                 proc_exit(0);
 548
 549                         default:
 550                                 ereport(FATAL,
 551                                                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
 552                                                  errmsg("invalid standby message type \"%c\"",
 553                                                                 firstchar)));
 554                 }
 555         }
 556
 557         /*
 558          * Save the last reply timestamp if we've received at least one reply.
 559          */
 560         if (received)
 561                 last_reply_timestamp = GetCurrentTimestamp();
 562 }
 563
 564 /*
 565  * Process a status update message received from standby.
 566  */
 567 static void
 568 ProcessStandbyMessage(void)
 569 {
 570         char            msgtype;
 571
 572         resetStringInfo(&reply_message);
 573
 574         /*
 575          * Read the message contents.
 576          */
 577         if (pq_getmessage(&reply_message, 0))
 578         {
 579                 ereport(COMMERROR,
 580                                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
 581                                  errmsg("unexpected EOF on standby connection")));
 582                 proc_exit(0);
 583         }
 584
 585         /*
 586          * Check message type from the first byte.
 587          */
 588         msgtype = pq_getmsgbyte(&reply_message);
 589
 590         switch (msgtype)
 591         {
 592                 case 'r':
 593                         ProcessStandbyReplyMessage();
 594                         break;
 595
 596                 case 'h':
 597                         ProcessStandbyHSFeedbackMessage();
 598                         break;
 599
 600                 default:
 601                         ereport(COMMERROR,
 602                                         (errcode(ERRCODE_PROTOCOL_VIOLATION),
 603                                          errmsg("unexpected message type \"%c\"", msgtype)));
 604                         proc_exit(0);
 605         }
 606 }
 607
 608 /*
 609  * Regular reply from standby advising of WAL positions on standby server.
 610  */
 611 static void
 612 ProcessStandbyReplyMessage(void)
 613 {
 614         StandbyReplyMessage reply;
 615
 616         pq_copymsgbytes(&reply_message, (char *) &reply, sizeof(StandbyReplyMessage));
 617
 618         elog(DEBUG2, "write %X/%X flush %X/%X apply %X/%X",
 619                  (uint32) (reply.write >> 32), (uint32) reply.write,
 620                  (uint32) (reply.flush >> 32), (uint32) reply.flush,
 621                  (uint32) (reply.apply >> 32), (uint32) reply.apply);
 622
 623         /*
 624          * Update shared state for this WalSender process based on reply data from
 625          * standby.
 626          */
 627         {
 628                 /* use volatile pointer to prevent code rearrangement */
 629                 volatile WalSnd *walsnd = MyWalSnd;
 630
 631                 SpinLockAcquire(&walsnd->mutex);
 632                 walsnd->write = reply.write;
 633                 walsnd->flush = reply.flush;
 634                 walsnd->apply = reply.apply;
 635                 SpinLockRelease(&walsnd->mutex);
 636         }
 637
 638         if (!am_cascading_walsender)
 639                 SyncRepReleaseWaiters();
 640 }
 641
 642 /*
 643  * Hot Standby feedback
 644  */
 645 static void
 646 ProcessStandbyHSFeedbackMessage(void)
 647 {
 648         StandbyHSFeedbackMessage reply;
 649         TransactionId nextXid;
 650         uint32          nextEpoch;
 651
 652         /* Decipher the reply message */
 653         pq_copymsgbytes(&reply_message, (char *) &reply,
 654                                         sizeof(StandbyHSFeedbackMessage));
 655
 656         elog(DEBUG2, "hot standby feedback xmin %u epoch %u",
 657                  reply.xmin,
 658                  reply.epoch);
 659
 660         /* Ignore invalid xmin (can't actually happen with current walreceiver) */
 661         if (!TransactionIdIsNormal(reply.xmin))
 662                 return;
 663
 664         /*
 665          * Check that the provided xmin/epoch are sane, that is, not in the future
 666          * and not so far back as to be already wrapped around.  Ignore if not.
 667          *
 668          * Epoch of nextXid should be same as standby, or if the counter has
 669          * wrapped, then one greater than standby.
 670          */
 671         GetNextXidAndEpoch(&nextXid, &nextEpoch);
 672
 673         if (reply.xmin <= nextXid)
 674         {
 675                 if (reply.epoch != nextEpoch)
 676                         return;
 677         }
 678         else
 679         {
 680                 if (reply.epoch + 1 != nextEpoch)
 681                         return;
 682         }
 683
 684         if (!TransactionIdPrecedesOrEquals(reply.xmin, nextXid))
 685                 return;                                 /* epoch OK, but it's wrapped around */
 686
 687         /*
 688          * Set the WalSender's xmin equal to the standby's requested xmin, so that
 689          * the xmin will be taken into account by GetOldestXmin.  This will hold
 690          * back the removal of dead rows and thereby prevent the generation of
 691          * cleanup conflicts on the standby server.
 692          *
 693          * There is a small window for a race condition here: although we just
 694          * checked that reply.xmin precedes nextXid, the nextXid could have gotten
 695          * advanced between our fetching it and applying the xmin below, perhaps
 696          * far enough to make reply.xmin wrap around.  In that case the xmin we
 697          * set here would be "in the future" and have no effect.  No point in
 698          * worrying about this since it's too late to save the desired data
 699          * anyway.      Assuming that the standby sends us an increasing sequence of
 700          * xmins, this could only happen during the first reply cycle, else our
 701          * own xmin would prevent nextXid from advancing so far.
 702          *
 703          * We don't bother taking the ProcArrayLock here.  Setting the xmin field
 704          * is assumed atomic, and there's no real need to prevent a concurrent
 705          * GetOldestXmin.  (If we're moving our xmin forward, this is obviously
 706          * safe, and if we're moving it backwards, well, the data is at risk
 707          * already since a VACUUM could have just finished calling GetOldestXmin.)
 708          */
 709         MyPgXact->xmin = reply.xmin;
 710 }
 711
 712 /* Main loop of walsender process */
 713 static void
 714 WalSndLoop(void)
 715 {
 716         char       *output_message;
 717         bool            caughtup = false;
 718
 719         /*
 720          * Allocate buffer that will be used for each output message.  We do this
 721          * just once to reduce palloc overhead.  The buffer must be made large
 722          * enough for maximum-sized messages.
 723          */
 724         output_message = palloc(1 + sizeof(WalDataMessageHeader) + MAX_SEND_SIZE);
 725
 726         /*
 727          * Allocate buffer that will be used for processing reply messages.  As
 728          * above, do this just once to reduce palloc overhead.
 729          */
 730         initStringInfo(&reply_message);
 731
 732         /* Initialize the last reply timestamp */
 733         last_reply_timestamp = GetCurrentTimestamp();
 734
 735         /* Loop forever, unless we get an error */
 736         for (;;)
 737         {
 738                 /* Clear any already-pending wakeups */
 739                 ResetLatch(&MyWalSnd->latch);
 740
 741                 /*
 742                  * Emergency bailout if postmaster has died.  This is to avoid the
 743                  * necessity for manual cleanup of all postmaster children.
 744                  */
 745                 if (!PostmasterIsAlive())
 746                         exit(1);
 747
 748                 /* Process any requests or signals received recently */
 749                 if (got_SIGHUP)
 750                 {
 751                         got_SIGHUP = false;
 752                         ProcessConfigFile(PGC_SIGHUP);
 753                         SyncRepInitConfig();
 754                 }
 755
 756                 /* Normal exit from the walsender is here */
 757                 if (walsender_shutdown_requested)
 758                 {
 759                         /* Inform the standby that XLOG streaming is done */
 760                         pq_puttextmessage('C', "COPY 0");
 761                         pq_flush();
 762
 763                         proc_exit(0);
 764                 }
 765
 766                 /* Check for input from the client */
 767                 ProcessRepliesIfAny();
 768
 769                 /*
 770                  * If we don't have any pending data in the output buffer, try to send
 771                  * some more.  If there is some, we don't bother to call XLogSend
 772                  * again until we've flushed it ... but we'd better assume we are not
 773                  * caught up.
 774                  */
 775                 if (!pq_is_send_pending())
 776                         XLogSend(output_message, &caughtup);
 777                 else
 778                         caughtup = false;
 779
 780                 /* Try to flush pending output to the client */
 781                 if (pq_flush_if_writable() != 0)
 782                         break;
 783
 784                 /* If nothing remains to be sent right now ... */
 785                 if (caughtup && !pq_is_send_pending())
 786                 {
 787                         /*
 788                          * If we're in catchup state, move to streaming.  This is an
 789                          * important state change for users to know about, since before
 790                          * this point data loss might occur if the primary dies and we
 791                          * need to failover to the standby. The state change is also
 792                          * important for synchronous replication, since commits that
 793                          * started to wait at that point might wait for some time.
 794                          */
 795                         if (MyWalSnd->state == WALSNDSTATE_CATCHUP)
 796                         {
 797                                 ereport(DEBUG1,
 798                                          (errmsg("standby \"%s\" has now caught up with primary",
 799                                                          application_name)));
 800                                 WalSndSetState(WALSNDSTATE_STREAMING);
 801                         }
 802
 803                         /*
 804                          * When SIGUSR2 arrives, we send any outstanding logs up to the
 805                          * shutdown checkpoint record (i.e., the latest record) and exit.
 806                          * This may be a normal termination at shutdown, or a promotion,
 807                          * the walsender is not sure which.
 808                          */
 809                         if (walsender_ready_to_stop)
 810                         {
 811                                 /* ... let's just be real sure we're caught up ... */
 812                                 XLogSend(output_message, &caughtup);
 813                                 if (caughtup && !pq_is_send_pending())
 814                                 {
 815                                         walsender_shutdown_requested = true;
 816                                         continue;       /* don't want to wait more */
 817                                 }
 818                         }
 819                 }
 820
 821                 /*
 822                  * We don't block if not caught up, unless there is unsent data
 823                  * pending in which case we'd better block until the socket is
 824                  * write-ready.  This test is only needed for the case where XLogSend
 825                  * loaded a subset of the available data but then pq_flush_if_writable
 826                  * flushed it all --- we should immediately try to send more.
 827                  */
 828                 if (caughtup || pq_is_send_pending())
 829                 {
 830                         TimestampTz timeout = 0;
 831                         long            sleeptime = 10000;              /* 10 s */
 832                         int                     wakeEvents;
 833
 834                         wakeEvents = WL_LATCH_SET | WL_POSTMASTER_DEATH |
 835                                 WL_SOCKET_READABLE | WL_TIMEOUT;
 836
 837                         if (pq_is_send_pending())
 838                                 wakeEvents |= WL_SOCKET_WRITEABLE;
 839                         else
 840                         {
 841                                 WalSndKeepalive(output_message);
 842                                 /* Try to flush pending output to the client */
 843                                 if (pq_flush_if_writable() != 0)
 844                                         break;
 845                         }
 846
 847                         /* Determine time until replication timeout */
 848                         if (replication_timeout > 0)
 849                         {
 850                                 timeout = TimestampTzPlusMilliseconds(last_reply_timestamp,
 851                                                                                                           replication_timeout);
 852                                 sleeptime = 1 + (replication_timeout / 10);
 853                         }
 854
 855                         /* Sleep until something happens or replication timeout */
 856                         WaitLatchOrSocket(&MyWalSnd->latch, wakeEvents,
 857                                                           MyProcPort->sock, sleeptime);
 858
 859                         /*
 860                          * Check for replication timeout.  Note we ignore the corner case
 861                          * possibility that the client replied just as we reached the
 862                          * timeout ... he's supposed to reply *before* that.
 863                          */
 864                         if (replication_timeout > 0 &&
 865                                 GetCurrentTimestamp() >= timeout)
 866                         {
 867                                 /*
 868                                  * Since typically expiration of replication timeout means
 869                                  * communication problem, we don't send the error message to
 870                                  * the standby.
 871                                  */
 872                                 ereport(COMMERROR,
 873                                                 (errmsg("terminating walsender process due to replication timeout")));
 874                                 break;
 875                         }
 876                 }
 877         }
 878
 879         /*
 880          * Get here on send failure.  Clean up and exit.
 881          *
 882          * Reset whereToSendOutput to prevent ereport from attempting to send any
 883          * more messages to the standby.
 884          */
 885         if (whereToSendOutput == DestRemote)
 886                 whereToSendOutput = DestNone;
 887
 888         proc_exit(0);
 889         abort();                                        /* keep the compiler quiet */
 890 }
 891
 892 /* Initialize a per-walsender data structure for this walsender process */
 893 static void
 894 InitWalSnd(void)
 895 {
 896         int                     i;
 897
 898         /*
 899          * WalSndCtl should be set up already (we inherit this by fork() or
 900          * EXEC_BACKEND mechanism from the postmaster).
 901          */
 902         Assert(WalSndCtl != NULL);
 903         Assert(MyWalSnd == NULL);
 904
 905         /*
 906          * Find a free walsender slot and reserve it. If this fails, we must be
 907          * out of WalSnd structures.
 908          */
 909         for (i = 0; i < max_wal_senders; i++)
 910         {
 911                 /* use volatile pointer to prevent code rearrangement */
 912                 volatile WalSnd *walsnd = &WalSndCtl->walsnds[i];
 913
 914                 SpinLockAcquire(&walsnd->mutex);
 915
 916                 if (walsnd->pid != 0)
 917                 {
 918                         SpinLockRelease(&walsnd->mutex);
 919                         continue;
 920                 }
 921                 else
 922                 {
 923                         /*
 924                          * Found a free slot. Reserve it for us.
 925                          */
 926                         walsnd->pid = MyProcPid;
 927                         MemSet(&walsnd->sentPtr, 0, sizeof(XLogRecPtr));
 928                         walsnd->state = WALSNDSTATE_STARTUP;
 929                         SpinLockRelease(&walsnd->mutex);
 930                         /* don't need the lock anymore */
 931                         OwnLatch((Latch *) &walsnd->latch);
 932                         MyWalSnd = (WalSnd *) walsnd;
 933
 934                         break;
 935                 }
 936         }
 937         if (MyWalSnd == NULL)
 938                 ereport(FATAL,
 939                                 (errcode(ERRCODE_TOO_MANY_CONNECTIONS),
 940                                  errmsg("number of requested standby connections "
 941                                                 "exceeds max_wal_senders (currently %d)",
 942                                                 max_wal_senders)));
 943
 944         /* Arrange to clean up at walsender exit */
 945         on_shmem_exit(WalSndKill, 0);
 946 }
 947
 948 /* Destroy the per-walsender data structure for this walsender process */
 949 static void
 950 WalSndKill(int code, Datum arg)
 951 {
 952         Assert(MyWalSnd != NULL);
 953
 954         /*
 955          * Mark WalSnd struct no longer in use. Assume that no lock is required
 956          * for this.
 957          */
 958         MyWalSnd->pid = 0;
 959         DisownLatch(&MyWalSnd->latch);
 960
 961         /* WalSnd struct isn't mine anymore */
 962         MyWalSnd = NULL;
 963 }
 964
 965 /*
 966  * Read 'count' bytes from WAL into 'buf', starting at location 'startptr'
 967  *
 968  * XXX probably this should be improved to suck data directly from the
 969  * WAL buffers when possible.
 970  *
 971  * Will open, and keep open, one WAL segment stored in the global file
 972  * descriptor sendFile. This means if XLogRead is used once, there will
 973  * always be one descriptor left open until the process ends, but never
 974  * more than one.
 975  */
 976 void
 977 XLogRead(char *buf, XLogRecPtr startptr, Size count)
 978 {
 979         char       *p;
 980         XLogRecPtr      recptr;
 981         Size            nbytes;
 982         XLogSegNo       lastRemovedSegNo;
 983         XLogSegNo       segno;
 984
 985 retry:
 986         p = buf;
 987         recptr = startptr;
 988         nbytes = count;
 989
 990         while (nbytes > 0)
 991         {
 992                 uint32          startoff;
 993                 int                     segbytes;
 994                 int                     readbytes;
 995
 996                 startoff = recptr % XLogSegSize;
 997
 998                 if (sendFile < 0 || !XLByteInSeg(recptr, sendSegNo))
 999                 {
1000                         char            path[MAXPGPATH];
1001
1002                         /* Switch to another logfile segment */
1003                         if (sendFile >= 0)
1004                                 close(sendFile);
1005
1006                         XLByteToSeg(recptr, sendSegNo);
1007                         XLogFilePath(path, ThisTimeLineID, sendSegNo);
1008
1009                         sendFile = BasicOpenFile(path, O_RDONLY | PG_BINARY, 0);
1010                         if (sendFile < 0)
1011                         {
1012                                 /*
1013                                  * If the file is not found, assume it's because the standby
1014                                  * asked for a too old WAL segment that has already been
1015                                  * removed or recycled.
1016                                  */
1017                                 if (errno == ENOENT)
1018                                         ereport(ERROR,
1019                                                         (errcode_for_file_access(),
1020                                                          errmsg("requested WAL segment %s has already been removed",
1021                                                                         XLogFileNameP(ThisTimeLineID, sendSegNo))));
1022                                 else
1023                                         ereport(ERROR,
1024                                                         (errcode_for_file_access(),
1025                                                          errmsg("could not open file \"%s\": %m",
1026                                                                         path)));
1027                         }
1028                         sendOff = 0;
1029                 }
1030
1031                 /* Need to seek in the file? */
1032                 if (sendOff != startoff)
1033                 {
1034                         if (lseek(sendFile, (off_t) startoff, SEEK_SET) < 0)
1035                                 ereport(ERROR,
1036                                                 (errcode_for_file_access(),
1037                                                  errmsg("could not seek in log segment %s to offset %u: %m",
1038                                                                 XLogFileNameP(ThisTimeLineID, sendSegNo),
1039                                                                 startoff)));
1040                         sendOff = startoff;
1041                 }
1042
1043                 /* How many bytes are within this segment? */
1044                 if (nbytes > (XLogSegSize - startoff))
1045                         segbytes = XLogSegSize - startoff;
1046                 else
1047                         segbytes = nbytes;
1048
1049                 readbytes = read(sendFile, p, segbytes);
1050                 if (readbytes <= 0)
1051                 {
1052                         ereport(ERROR,
1053                                         (errcode_for_file_access(),
1054                         errmsg("could not read from log segment %s, offset %u, length %lu: %m",
1055                                    XLogFileNameP(ThisTimeLineID, sendSegNo),
1056                                    sendOff, (unsigned long) segbytes)));
1057                 }
1058
1059                 /* Update state for read */
1060                 XLByteAdvance(recptr, readbytes);
1061
1062                 sendOff += readbytes;
1063                 nbytes -= readbytes;
1064                 p += readbytes;
1065         }
1066
1067         /*
1068          * After reading into the buffer, check that what we read was valid. We do
1069          * this after reading, because even though the segment was present when we
1070          * opened it, it might get recycled or removed while we read it. The
1071          * read() succeeds in that case, but the data we tried to read might
1072          * already have been overwritten with new WAL records.
1073          */
1074         XLogGetLastRemoved(&lastRemovedSegNo);
1075         XLByteToSeg(startptr, segno);
1076         if (segno <= lastRemovedSegNo)
1077                 ereport(ERROR,
1078                                 (errcode_for_file_access(),
1079                                  errmsg("requested WAL segment %s has already been removed",
1080                                                 XLogFileNameP(ThisTimeLineID, segno))));
1081
1082         /*
1083          * During recovery, the currently-open WAL file might be replaced with the
1084          * file of the same name retrieved from archive. So we always need to
1085          * check what we read was valid after reading into the buffer. If it's
1086          * invalid, we try to open and read the file again.
1087          */
1088         if (am_cascading_walsender)
1089         {
1090                 /* use volatile pointer to prevent code rearrangement */
1091                 volatile WalSnd *walsnd = MyWalSnd;
1092                 bool            reload;
1093
1094                 SpinLockAcquire(&walsnd->mutex);
1095                 reload = walsnd->needreload;
1096                 walsnd->needreload = false;
1097                 SpinLockRelease(&walsnd->mutex);
1098
1099                 if (reload && sendFile >= 0)
1100                 {
1101                         close(sendFile);
1102                         sendFile = -1;
1103
1104                         goto retry;
1105                 }
1106         }
1107 }
1108
1109 /*
1110  * Read up to MAX_SEND_SIZE bytes of WAL that's been flushed to disk,
1111  * but not yet sent to the client, and buffer it in the libpq output
1112  * buffer.
1113  *
1114  * msgbuf is a work area in which the output message is constructed.  It's
1115  * passed in just so we can avoid re-palloc'ing the buffer on each cycle.
1116  * It must be of size 1 + sizeof(WalDataMessageHeader) + MAX_SEND_SIZE.
1117  *
1118  * If there is no unsent WAL remaining, *caughtup is set to true, otherwise
1119  * *caughtup is set to false.
1120
1121  */
1122 static void
1123 XLogSend(char *msgbuf, bool *caughtup)
1124 {
1125         XLogRecPtr      SendRqstPtr;
1126         XLogRecPtr      startptr;
1127         XLogRecPtr      endptr;
1128         Size            nbytes;
1129         WalDataMessageHeader msghdr;
1130
1131         /*
1132          * Attempt to send all data that's already been written out and fsync'd to
1133          * disk.  We cannot go further than what's been written out given the
1134          * current implementation of XLogRead().  And in any case it's unsafe to
1135          * send WAL that is not securely down to disk on the master: if the master
1136          * subsequently crashes and restarts, slaves must not have applied any WAL
1137          * that gets lost on the master.
1138          */
1139         SendRqstPtr = am_cascading_walsender ? GetStandbyFlushRecPtr() : GetFlushRecPtr();
1140
1141         /* Quick exit if nothing to do */
1142         if (XLByteLE(SendRqstPtr, sentPtr))
1143         {
1144                 *caughtup = true;
1145                 return;
1146         }
1147
1148         /*
1149          * Figure out how much to send in one message. If there's no more than
1150          * MAX_SEND_SIZE bytes to send, send everything. Otherwise send
1151          * MAX_SEND_SIZE bytes, but round back to logfile or page boundary.
1152          *
1153          * The rounding is not only for performance reasons. Walreceiver relies on
1154          * the fact that we never split a WAL record across two messages. Since a
1155          * long WAL record is split at page boundary into continuation records,
1156          * page boundary is always a safe cut-off point. We also assume that
1157          * SendRqstPtr never points to the middle of a WAL record.
1158          */
1159         startptr = sentPtr;
1160         endptr = startptr;
1161         XLByteAdvance(endptr, MAX_SEND_SIZE);
1162
1163         /* if we went beyond SendRqstPtr, back off */
1164         if (XLByteLE(SendRqstPtr, endptr))
1165         {
1166                 endptr = SendRqstPtr;
1167                 *caughtup = true;
1168         }
1169         else
1170         {
1171                 /* round down to page boundary. */
1172                 endptr -= (endptr % XLOG_BLCKSZ);
1173                 *caughtup = false;
1174         }
1175
1176         nbytes = endptr - startptr;
1177         Assert(nbytes <= MAX_SEND_SIZE);
1178
1179         /*
1180          * OK to read and send the slice.
1181          */
1182         msgbuf[0] = 'w';
1183
1184         /*
1185          * Read the log directly into the output buffer to avoid extra memcpy
1186          * calls.
1187          */
1188         XLogRead(msgbuf + 1 + sizeof(WalDataMessageHeader), startptr, nbytes);
1189
1190         /*
1191          * We fill the message header last so that the send timestamp is taken as
1192          * late as possible.
1193          */
1194         msghdr.dataStart = startptr;
1195         msghdr.walEnd = SendRqstPtr;
1196         msghdr.sendTime = GetCurrentTimestamp();
1197
1198         memcpy(msgbuf + 1, &msghdr, sizeof(WalDataMessageHeader));
1199
1200         pq_putmessage_noblock('d', msgbuf, 1 + sizeof(WalDataMessageHeader) + nbytes);
1201
1202         sentPtr = endptr;
1203
1204         /* Update shared memory status */
1205         {
1206                 /* use volatile pointer to prevent code rearrangement */
1207                 volatile WalSnd *walsnd = MyWalSnd;
1208
1209                 SpinLockAcquire(&walsnd->mutex);
1210                 walsnd->sentPtr = sentPtr;
1211                 SpinLockRelease(&walsnd->mutex);
1212         }
1213
1214         /* Report progress of XLOG streaming in PS display */
1215         if (update_process_title)
1216         {
1217                 char            activitymsg[50];
1218
1219                 snprintf(activitymsg, sizeof(activitymsg), "streaming %X/%X",
1220                                  (uint32) (sentPtr >> 32), (uint32) sentPtr);
1221                 set_ps_display(activitymsg, false);
1222         }
1223
1224         return;
1225 }
1226
1227 /*
1228  * Request walsenders to reload the currently-open WAL file
1229  */
1230 void
1231 WalSndRqstFileReload(void)
1232 {
1233         int                     i;
1234
1235         for (i = 0; i < max_wal_senders; i++)
1236         {
1237                 /* use volatile pointer to prevent code rearrangement */
1238                 volatile WalSnd *walsnd = &WalSndCtl->walsnds[i];
1239
1240                 if (walsnd->pid == 0)
1241                         continue;
1242
1243                 SpinLockAcquire(&walsnd->mutex);
1244                 walsnd->needreload = true;
1245                 SpinLockRelease(&walsnd->mutex);
1246         }
1247 }
1248
1249 /* SIGHUP: set flag to re-read config file at next convenient time */
1250 static void
1251 WalSndSigHupHandler(SIGNAL_ARGS)
1252 {
1253         int                     save_errno = errno;
1254
1255         got_SIGHUP = true;
1256         if (MyWalSnd)
1257                 SetLatch(&MyWalSnd->latch);
1258
1259         errno = save_errno;
1260 }
1261
1262 /* SIGTERM: set flag to shut down */
1263 static void
1264 WalSndShutdownHandler(SIGNAL_ARGS)
1265 {
1266         int                     save_errno = errno;
1267
1268         walsender_shutdown_requested = true;
1269         if (MyWalSnd)
1270                 SetLatch(&MyWalSnd->latch);
1271
1272         /*
1273          * Set the standard (non-walsender) state as well, so that we can abort
1274          * things like do_pg_stop_backup().
1275          */
1276         InterruptPending = true;
1277         ProcDiePending = true;
1278
1279         errno = save_errno;
1280 }
1281
1282 /*
1283  * WalSndQuickDieHandler() occurs when signalled SIGQUIT by the postmaster.
1284  *
1285  * Some backend has bought the farm,
1286  * so we need to stop what we're doing and exit.
1287  */
1288 static void
1289 WalSndQuickDieHandler(SIGNAL_ARGS)
1290 {
1291         PG_SETMASK(&BlockSig);
1292
1293         /*
1294          * We DO NOT want to run proc_exit() callbacks -- we're here because
1295          * shared memory may be corrupted, so we don't want to try to clean up our
1296          * transaction.  Just nail the windows shut and get out of town.  Now that
1297          * there's an atexit callback to prevent third-party code from breaking
1298          * things by calling exit() directly, we have to reset the callbacks
1299          * explicitly to make this work as intended.
1300          */
1301         on_exit_reset();
1302
1303         /*
1304          * Note we do exit(2) not exit(0).      This is to force the postmaster into a
1305          * system reset cycle if some idiot DBA sends a manual SIGQUIT to a random
1306          * backend.  This is necessary precisely because we don't clean up our
1307          * shared memory state.  (The "dead man switch" mechanism in pmsignal.c
1308          * should ensure the postmaster sees this as a crash, too, but no harm in
1309          * being doubly sure.)
1310          */
1311         exit(2);
1312 }
1313
1314 /* SIGUSR1: set flag to send WAL records */
1315 static void
1316 WalSndXLogSendHandler(SIGNAL_ARGS)
1317 {
1318         int                     save_errno = errno;
1319
1320         latch_sigusr1_handler();
1321
1322         errno = save_errno;
1323 }
1324
1325 /* SIGUSR2: set flag to do a last cycle and shut down afterwards */
1326 static void
1327 WalSndLastCycleHandler(SIGNAL_ARGS)
1328 {
1329         int                     save_errno = errno;
1330
1331         walsender_ready_to_stop = true;
1332         if (MyWalSnd)
1333                 SetLatch(&MyWalSnd->latch);
1334
1335         errno = save_errno;
1336 }
1337
1338 /* Set up signal handlers */
1339 void
1340 WalSndSignals(void)
1341 {
1342         /* Set up signal handlers */
1343         pqsignal(SIGHUP, WalSndSigHupHandler);          /* set flag to read config
1344                                                                                                  * file */
1345         pqsignal(SIGINT, SIG_IGN);      /* not used */
1346         pqsignal(SIGTERM, WalSndShutdownHandler);       /* request shutdown */
1347         pqsignal(SIGQUIT, WalSndQuickDieHandler);       /* hard crash time */
1348         pqsignal(SIGALRM, SIG_IGN);
1349         pqsignal(SIGPIPE, SIG_IGN);
1350         pqsignal(SIGUSR1, WalSndXLogSendHandler);       /* request WAL sending */
1351         pqsignal(SIGUSR2, WalSndLastCycleHandler);      /* request a last cycle and
1352                                                                                                  * shutdown */
1353
1354         /* Reset some signals that are accepted by postmaster but not here */
1355         pqsignal(SIGCHLD, SIG_DFL);
1356         pqsignal(SIGTTIN, SIG_DFL);
1357         pqsignal(SIGTTOU, SIG_DFL);
1358         pqsignal(SIGCONT, SIG_DFL);
1359         pqsignal(SIGWINCH, SIG_DFL);
1360 }
1361
1362 /* Report shared-memory space needed by WalSndShmemInit */
1363 Size
1364 WalSndShmemSize(void)
1365 {
1366         Size            size = 0;
1367
1368         size = offsetof(WalSndCtlData, walsnds);
1369         size = add_size(size, mul_size(max_wal_senders, sizeof(WalSnd)));
1370
1371         return size;
1372 }
1373
1374 /* Allocate and initialize walsender-related shared memory */
1375 void
1376 WalSndShmemInit(void)
1377 {
1378         bool            found;
1379         int                     i;
1380
1381         WalSndCtl = (WalSndCtlData *)
1382                 ShmemInitStruct("Wal Sender Ctl", WalSndShmemSize(), &found);
1383
1384         if (!found)
1385         {
1386                 /* First time through, so initialize */
1387                 MemSet(WalSndCtl, 0, WalSndShmemSize());
1388
1389                 for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++)
1390                         SHMQueueInit(&(WalSndCtl->SyncRepQueue[i]));
1391
1392                 for (i = 0; i < max_wal_senders; i++)
1393                 {
1394                         WalSnd     *walsnd = &WalSndCtl->walsnds[i];
1395
1396                         SpinLockInit(&walsnd->mutex);
1397                         InitSharedLatch(&walsnd->latch);
1398                 }
1399         }
1400 }
1401
1402 /*
1403  * Wake up all walsenders
1404  *
1405  * This will be called inside critical sections, so throwing an error is not
1406  * adviseable.
1407  */
1408 void
1409 WalSndWakeup(void)
1410 {
1411         int                     i;
1412
1413         for (i = 0; i < max_wal_senders; i++)
1414                 SetLatch(&WalSndCtl->walsnds[i].latch);
1415 }
1416
1417 /* Set state for current walsender (only called in walsender) */
1418 void
1419 WalSndSetState(WalSndState state)
1420 {
1421         /* use volatile pointer to prevent code rearrangement */
1422         volatile WalSnd *walsnd = MyWalSnd;
1423
1424         Assert(am_walsender);
1425
1426         if (walsnd->state == state)
1427                 return;
1428
1429         SpinLockAcquire(&walsnd->mutex);
1430         walsnd->state = state;
1431         SpinLockRelease(&walsnd->mutex);
1432 }
1433
1434 /*
1435  * Return a string constant representing the state. This is used
1436  * in system views, and should *not* be translated.
1437  */
1438 static const char *
1439 WalSndGetStateString(WalSndState state)
1440 {
1441         switch (state)
1442         {
1443                 case WALSNDSTATE_STARTUP:
1444                         return "startup";
1445                 case WALSNDSTATE_BACKUP:
1446                         return "backup";
1447                 case WALSNDSTATE_CATCHUP:
1448                         return "catchup";
1449                 case WALSNDSTATE_STREAMING:
1450                         return "streaming";
1451         }
1452         return "UNKNOWN";
1453 }
1454
1455
1456 /*
1457  * Returns activity of walsenders, including pids and xlog locations sent to
1458  * standby servers.
1459  */
1460 Datum
1461 pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
1462 {
1463 #define PG_STAT_GET_WAL_SENDERS_COLS    8
1464         ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
1465         TupleDesc       tupdesc;
1466         Tuplestorestate *tupstore;
1467         MemoryContext per_query_ctx;
1468         MemoryContext oldcontext;
1469         int                *sync_priority;
1470         int                     priority = 0;
1471         int                     sync_standby = -1;
1472         int                     i;
1473
1474         /* check to see if caller supports us returning a tuplestore */
1475         if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
1476                 ereport(ERROR,
1477                                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1478                                  errmsg("set-valued function called in context that cannot accept a set")));
1479         if (!(rsinfo->allowedModes & SFRM_Materialize))
1480                 ereport(ERROR,
1481                                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1482                                  errmsg("materialize mode required, but it is not " \
1483                                                 "allowed in this context")));
1484
1485         /* Build a tuple descriptor for our result type */
1486         if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
1487                 elog(ERROR, "return type must be a row type");
1488
1489         per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
1490         oldcontext = MemoryContextSwitchTo(per_query_ctx);
1491
1492         tupstore = tuplestore_begin_heap(true, false, work_mem);
1493         rsinfo->returnMode = SFRM_Materialize;
1494         rsinfo->setResult = tupstore;
1495         rsinfo->setDesc = tupdesc;
1496
1497         MemoryContextSwitchTo(oldcontext);
1498
1499         /*
1500          * Get the priorities of sync standbys all in one go, to minimise lock
1501          * acquisitions and to allow us to evaluate who is the current sync
1502          * standby. This code must match the code in SyncRepReleaseWaiters().
1503          */
1504         sync_priority = palloc(sizeof(int) * max_wal_senders);
1505         LWLockAcquire(SyncRepLock, LW_SHARED);
1506         for (i = 0; i < max_wal_senders; i++)
1507         {
1508                 /* use volatile pointer to prevent code rearrangement */
1509                 volatile WalSnd *walsnd = &WalSndCtl->walsnds[i];
1510
1511                 if (walsnd->pid != 0)
1512                 {
1513                         sync_priority[i] = walsnd->sync_standby_priority;
1514
1515                         if (walsnd->state == WALSNDSTATE_STREAMING &&
1516                                 walsnd->sync_standby_priority > 0 &&
1517                                 (priority == 0 ||
1518                                  priority > walsnd->sync_standby_priority))
1519                         {
1520                                 priority = walsnd->sync_standby_priority;
1521                                 sync_standby = i;
1522                         }
1523                 }
1524         }
1525         LWLockRelease(SyncRepLock);
1526
1527         for (i = 0; i < max_wal_senders; i++)
1528         {
1529                 /* use volatile pointer to prevent code rearrangement */
1530                 volatile WalSnd *walsnd = &WalSndCtl->walsnds[i];
1531                 char            location[MAXFNAMELEN];
1532                 XLogRecPtr      sentPtr;
1533                 XLogRecPtr      write;
1534                 XLogRecPtr      flush;
1535                 XLogRecPtr      apply;
1536                 WalSndState state;
1537                 Datum           values[PG_STAT_GET_WAL_SENDERS_COLS];
1538                 bool            nulls[PG_STAT_GET_WAL_SENDERS_COLS];
1539
1540                 if (walsnd->pid == 0)
1541                         continue;
1542
1543                 SpinLockAcquire(&walsnd->mutex);
1544                 sentPtr = walsnd->sentPtr;
1545                 state = walsnd->state;
1546                 write = walsnd->write;
1547                 flush = walsnd->flush;
1548                 apply = walsnd->apply;
1549                 SpinLockRelease(&walsnd->mutex);
1550
1551                 memset(nulls, 0, sizeof(nulls));
1552                 values[0] = Int32GetDatum(walsnd->pid);
1553
1554                 if (!superuser())
1555                 {
1556                         /*
1557                          * Only superusers can see details. Other users only get the pid
1558                          * value to know it's a walsender, but no details.
1559                          */
1560                         MemSet(&nulls[1], true, PG_STAT_GET_WAL_SENDERS_COLS - 1);
1561                 }
1562                 else
1563                 {
1564                         values[1] = CStringGetTextDatum(WalSndGetStateString(state));
1565
1566                         snprintf(location, sizeof(location), "%X/%X",
1567                                          (uint32) (sentPtr >> 32), (uint32) sentPtr);
1568                         values[2] = CStringGetTextDatum(location);
1569
1570                         if (write == 0)
1571                                 nulls[3] = true;
1572                         snprintf(location, sizeof(location), "%X/%X",
1573                                          (uint32) (write >> 32), (uint32) write);
1574                         values[3] = CStringGetTextDatum(location);
1575
1576                         if (flush == 0)
1577                                 nulls[4] = true;
1578                         snprintf(location, sizeof(location), "%X/%X",
1579                                          (uint32) (flush >> 32), (uint32) flush);
1580                         values[4] = CStringGetTextDatum(location);
1581
1582                         if (apply == 0)
1583                                 nulls[5] = true;
1584                         snprintf(location, sizeof(location), "%X/%X",
1585                                          (uint32) (apply >> 32), (uint32) apply);
1586                         values[5] = CStringGetTextDatum(location);
1587
1588                         values[6] = Int32GetDatum(sync_priority[i]);
1589
1590                         /*
1591                          * More easily understood version of standby state. This is purely
1592                          * informational, not different from priority.
1593                          */
1594                         if (sync_priority[i] == 0)
1595                                 values[7] = CStringGetTextDatum("async");
1596                         else if (i == sync_standby)
1597                                 values[7] = CStringGetTextDatum("sync");
1598                         else
1599                                 values[7] = CStringGetTextDatum("potential");
1600                 }
1601
1602                 tuplestore_putvalues(tupstore, tupdesc, values, nulls);
1603         }
1604         pfree(sync_priority);
1605
1606         /* clean up and return the tuplestore */
1607         tuplestore_donestoring(tupstore);
1608
1609         return (Datum) 0;
1610 }
1611
1612 static void
1613 WalSndKeepalive(char *msgbuf)
1614 {
1615         PrimaryKeepaliveMessage keepalive_message;
1616
1617         /* Construct a new message */
1618         keepalive_message.walEnd = sentPtr;
1619         keepalive_message.sendTime = GetCurrentTimestamp();
1620
1621         elog(DEBUG2, "sending replication keepalive");
1622
1623         /* Prepend with the message type and send it. */
1624         msgbuf[0] = 'k';
1625         memcpy(msgbuf + 1, &keepalive_message, sizeof(PrimaryKeepaliveMessage));
1626         pq_putmessage_noblock('d', msgbuf, sizeof(PrimaryKeepaliveMessage) + 1);
1627 }
1628
1629 /*
1630  * This isn't currently used for anything. Monitoring tools might be
1631  * interested in the future, and we'll need something like this in the
1632  * future for synchronous replication.
1633  */
1634 #ifdef NOT_USED
1635 /*
1636  * Returns the oldest Send position among walsenders. Or InvalidXLogRecPtr
1637  * if none.
1638  */
1639 XLogRecPtr
1640 GetOldestWALSendPointer(void)
1641 {
1642         XLogRecPtr      oldest = {0, 0};
1643         int                     i;
1644         bool            found = false;
1645
1646         for (i = 0; i < max_wal_senders; i++)
1647         {
1648                 /* use volatile pointer to prevent code rearrangement */
1649                 volatile WalSnd *walsnd = &WalSndCtl->walsnds[i];
1650                 XLogRecPtr      recptr;
1651
1652                 if (walsnd->pid == 0)
1653                         continue;
1654
1655                 SpinLockAcquire(&walsnd->mutex);
1656                 recptr = walsnd->sentPtr;
1657                 SpinLockRelease(&walsnd->mutex);
1658
1659                 if (recptr.xlogid == 0 && recptr.xrecoff == 0)
1660                         continue;
1661
1662                 if (!found || XLByteLT(recptr, oldest))
1663                         oldest = recptr;
1664                 found = true;
1665         }
1666         return oldest;
1667 }
1668
1669 #endif