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