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