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