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