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