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