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