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