granicus.if.org Git - postgresql/blob - src/backend/replication/walsender.c

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