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