*** a/doc/src/sgml/protocol.sgml --- b/doc/src/sgml/protocol.sgml *************** *** 1360,1365 **** The commands accepted in walsender mode are: --- 1360,1400 ---- + XLogRecPtr (F) + + + + + + + Byte1('l') + + + + Identifies the message as an acknowledgment of replication. + + + + + + Byte8 + + + + The end of the WAL data replicated to the standby, given in + XLogRecPtr format. + + + + + + + + + + + + XLogData (B) *** a/src/backend/Makefile --- b/src/backend/Makefile *************** *** 208,213 **** endif --- 208,214 ---- $(INSTALL_DATA) $(srcdir)/libpq/pg_ident.conf.sample '$(DESTDIR)$(datadir)/pg_ident.conf.sample' $(INSTALL_DATA) $(srcdir)/utils/misc/postgresql.conf.sample '$(DESTDIR)$(datadir)/postgresql.conf.sample' $(INSTALL_DATA) $(srcdir)/access/transam/recovery.conf.sample '$(DESTDIR)$(datadir)/recovery.conf.sample' + $(INSTALL_DATA) $(srcdir)/replication/standbys.conf.sample '$(DESTDIR)$(datadir)/standbys.conf.sample' install-bin: postgres $(POSTGRES_IMP) installdirs $(INSTALL_PROGRAM) postgres$(X) '$(DESTDIR)$(bindir)/postgres$(X)' *************** *** 262,268 **** endif rm -f '$(DESTDIR)$(datadir)/pg_hba.conf.sample' \ '$(DESTDIR)$(datadir)/pg_ident.conf.sample' \ '$(DESTDIR)$(datadir)/postgresql.conf.sample' \ ! '$(DESTDIR)$(datadir)/recovery.conf.sample' ########################################################################## --- 263,270 ---- rm -f '$(DESTDIR)$(datadir)/pg_hba.conf.sample' \ '$(DESTDIR)$(datadir)/pg_ident.conf.sample' \ '$(DESTDIR)$(datadir)/postgresql.conf.sample' \ ! '$(DESTDIR)$(datadir)/recovery.conf.sample' \ ! '$(DESTDIR)$(datadir)/standbys.conf.sample' ########################################################################## *** a/src/backend/access/transam/recovery.conf.sample --- b/src/backend/access/transam/recovery.conf.sample *************** *** 91,102 **** #--------------------------------------------------------------------------- # # When standby_mode is enabled, the PostgreSQL server will work as ! # a standby. It tries to connect to the primary according to the ! # connection settings primary_conninfo, and receives XLOG records ! # continuously. # #standby_mode = 'off' # #primary_conninfo = '' # e.g. 'host=localhost port=5432' # # --- 91,104 ---- #--------------------------------------------------------------------------- # # When standby_mode is enabled, the PostgreSQL server will work as ! # a standby under the name of standby_name. It tries to connect to ! # the primary according to the connection settings primary_conninfo, ! # and receives XLOG records continuously. # #standby_mode = 'off' # + #standby_name = '' + # #primary_conninfo = '' # e.g. 'host=localhost port=5432' # # *** a/src/backend/access/transam/twophase.c --- b/src/backend/access/transam/twophase.c *************** *** 55,60 **** --- 55,61 ---- #include "miscadmin.h" #include "pg_trace.h" #include "pgstat.h" + #include "replication/walsender.h" #include "storage/fd.h" #include "storage/procarray.h" #include "storage/sinvaladt.h" *************** *** 1062,1067 **** EndPrepare(GlobalTransaction gxact) --- 1063,1080 ---- END_CRIT_SECTION(); + /* + * Wait for WAL to be replicated up to the PREPARE record + * if replication is enabled. This operation has to be performed + * after the PREPARE record is generated and before other + * transactions know that this one has already been prepared. + * + * XXX: Since the caller prevents cancel/die interrupt, we cannot + * process that while waiting. Should we remove this restriction? + */ + if (max_wal_senders > 0) + WaitXLogSend(gxact->prepare_lsn); + records.tail = records.head = NULL; } *************** *** 2012,2017 **** RecordTransactionCommitPrepared(TransactionId xid, --- 2025,2039 ---- MyProc->inCommit = false; END_CRIT_SECTION(); + + /* + * Wait for WAL to be replicated up to the COMMIT PREPARED record + * if replication is enabled. This operation has to be performed + * after the COMMIT PREPARED record is generated and before other + * transactions know that this one has already been committed. + */ + if (max_wal_senders > 0) + WaitXLogSend(recptr); } /* *************** *** 2084,2087 **** RecordTransactionAbortPrepared(TransactionId xid, --- 2106,2118 ---- TransactionIdAbortTree(xid, nchildren, children); END_CRIT_SECTION(); + + /* + * Wait for WAL to be replicated up to the ABORT PREPARED record + * if replication is enabled. This operation has to be performed + * after the ABORT PREPARED record is generated and before other + * transactions know that this one has already been aborted. + */ + if (max_wal_senders > 0) + WaitXLogSend(recptr); } *** a/src/backend/access/transam/xact.c --- b/src/backend/access/transam/xact.c *************** *** 36,41 **** --- 36,42 ---- #include "libpq/be-fsstubs.h" #include "miscadmin.h" #include "pgstat.h" + #include "replication/walsender.h" #include "storage/bufmgr.h" #include "storage/fd.h" #include "storage/lmgr.h" *************** *** 1110,1115 **** RecordTransactionCommit(void) --- 1111,1128 ---- /* Compute latestXid while we have the child XIDs handy */ latestXid = TransactionIdLatest(xid, nchildren, children); + /* + * Wait for WAL to be replicated up to the COMMIT record if replication + * is enabled. This operation has to be performed after the COMMIT record + * is generated and before other transactions know that this one has + * already been committed. + * + * XXX: Since the caller prevents cancel/die interrupt, we cannot + * process that while waiting. Should we remove this restriction? + */ + if (max_wal_senders > 0) + WaitXLogSend(XactLastRecEnd); + /* Reset XactLastRecEnd until the next transaction writes something */ XactLastRecEnd.xrecoff = 0; *** a/src/backend/access/transam/xlog.c --- b/src/backend/access/transam/xlog.c *************** *** 189,194 **** static TimestampTz recoveryTargetTime; --- 189,195 ---- static bool StandbyMode = false; static char *PrimaryConnInfo = NULL; static char *TriggerFile = NULL; + static char *StandbyName = NULL; /* if recoveryStopsHere returns true, it saves actual stop xid/time here */ static TransactionId recoveryStopXid; *************** *** 532,537 **** typedef struct xl_parameter_change --- 533,548 ---- int wal_level; } xl_parameter_change; + /* Replication mode names */ + const char *ReplicationModeNames[] = { + "async", /* REPLICATION_MODE_ASYNC */ + "recv", /* REPLICATION_MODE_RECV */ + "fsync", /* REPLICATION_MODE_FSYNC */ + "replay" /* REPLICATION_MODE_REPLAY */ + }; + + ReplicationMode rplMode = InvalidReplicationMode; + /* * Flags set by interrupt handlers for later service in the redo loop. */ *************** *** 5258,5263 **** readRecoveryCommandFile(void) --- 5269,5281 ---- (errmsg("trigger_file = '%s'", TriggerFile))); } + else if (strcmp(tok1, "standby_name") == 0) + { + StandbyName = pstrdup(tok2); + ereport(DEBUG2, + (errmsg("standby_name = '%s'", + StandbyName))); + } else ereport(FATAL, (errmsg("unrecognized recovery parameter \"%s\"", *************** *** 6867,6872 **** GetFlushRecPtr(void) --- 6885,6907 ---- } /* + * GetReplayRecPtr -- Returns the last replay position. + */ + XLogRecPtr + GetReplayRecPtr(void) + { + /* use volatile pointer to prevent code rearrangement */ + volatile XLogCtlData *xlogctl = XLogCtl; + XLogRecPtr recptr; + + SpinLockAcquire(&xlogctl->info_lck); + recptr = xlogctl->recoveryLastRecPtr; + SpinLockRelease(&xlogctl->info_lck); + + return recptr; + } + + /* * Get the time of the last xlog segment switch */ pg_time_t *************** *** 8828,8842 **** pg_last_xlog_receive_location(PG_FUNCTION_ARGS) Datum pg_last_xlog_replay_location(PG_FUNCTION_ARGS) { - /* use volatile pointer to prevent code rearrangement */ - volatile XLogCtlData *xlogctl = XLogCtl; XLogRecPtr recptr; char location[MAXFNAMELEN]; ! SpinLockAcquire(&xlogctl->info_lck); ! recptr = xlogctl->recoveryLastRecPtr; ! SpinLockRelease(&xlogctl->info_lck); ! if (recptr.xlogid == 0 && recptr.xrecoff == 0) PG_RETURN_NULL(); --- 8863,8872 ---- Datum pg_last_xlog_replay_location(PG_FUNCTION_ARGS) { XLogRecPtr recptr; char location[MAXFNAMELEN]; ! recptr = GetReplayRecPtr(); if (recptr.xlogid == 0 && recptr.xrecoff == 0) PG_RETURN_NULL(); *************** *** 9467,9473 **** retry: { RequestXLogStreaming( fetching_ckpt ? RedoStartLSN : *RecPtr, ! PrimaryConnInfo); continue; } } --- 9497,9503 ---- { RequestXLogStreaming( fetching_ckpt ? RedoStartLSN : *RecPtr, ! PrimaryConnInfo, StandbyName); continue; } } *************** *** 9681,9683 **** CheckForStandbyTrigger(void) --- 9711,9727 ---- } return false; } + + /* + * Look up replication mode value by name. + */ + ReplicationMode + ReplicationModeNameGetValue(char *name) + { + ReplicationMode mode; + + for (mode = 0; mode <= MAXREPLICATIONMODE; mode++) + if (strcmp(ReplicationModeNames[mode], name) == 0) + return mode; + return InvalidReplicationMode; + } *** a/src/backend/libpq/hba.c --- b/src/backend/libpq/hba.c *************** *** 38,46 **** #define atooid(x) ((Oid) strtoul((x), NULL, 10)) #define atoxid(x) ((TransactionId) strtoul((x), NULL, 10)) - /* This is used to separate values in multi-valued column strings */ - #define MULTI_VALUE_SEP "\001" - #define MAX_TOKEN 256 /* callback data for check_network_callback */ --- 38,43 ---- *************** *** 54,59 **** typedef struct check_network_data --- 51,59 ---- /* pre-parsed content of HBA config file: list of HbaLine structs */ static List *parsed_hba_lines = NIL; + static const char *hba_keywords[] = {"all", "sameuser", "samegroup", "samerole", + "replication", NULL}; + /* * These variables hold the pre-parsed contents of the ident usermap * configuration file. ident_lines is a list of sublists, one sublist for *************** *** 67,76 **** static List *ident_lines = NIL; static List *ident_line_nums = NIL; - static void tokenize_file(const char *filename, FILE *file, - List **lines, List **line_nums); static char *tokenize_inc_file(const char *outer_filename, ! const char *inc_filename); /* * isblank() exists in the ISO C99 spec, but it's not very portable yet, --- 67,74 ---- static List *ident_line_nums = NIL; static char *tokenize_inc_file(const char *outer_filename, ! const char *inc_filename, const char **keywords); /* * isblank() exists in the ISO C99 spec, but it's not very portable yet, *************** *** 108,114 **** pg_isblank(const char c) * token. */ static bool ! next_token(FILE *fp, char *buf, int bufsz, bool *initial_quote) { int c; char *start_buf = buf; --- 106,113 ---- * token. */ static bool ! next_token(const char *filename, FILE *fp, char *buf, int bufsz, ! bool *initial_quote, const char **keywords) { int c; char *start_buf = buf; *************** *** 155,162 **** next_token(FILE *fp, char *buf, int bufsz, bool *initial_quote) *buf = '\0'; ereport(LOG, (errcode(ERRCODE_CONFIG_FILE_ERROR), ! errmsg("authentication file token too long, skipping: \"%s\"", ! start_buf))); /* Discard remainder of line */ while ((c = getc(fp)) != EOF && c != '\n') ; --- 154,161 ---- *buf = '\0'; ereport(LOG, (errcode(ERRCODE_CONFIG_FILE_ERROR), ! errmsg("configuration file \"%s\" token too long, skipping: \"%s\"", ! filename, start_buf))); /* Discard remainder of line */ while ((c = getc(fp)) != EOF && c != '\n') ; *************** *** 196,211 **** next_token(FILE *fp, char *buf, int bufsz, bool *initial_quote) *buf = '\0'; ! if (!saw_quote && ! (strcmp(start_buf, "all") == 0 || ! strcmp(start_buf, "sameuser") == 0 || ! strcmp(start_buf, "samegroup") == 0 || ! strcmp(start_buf, "samerole") == 0 || ! strcmp(start_buf, "replication") == 0)) { ! /* append newline to a magical keyword */ ! *buf++ = '\n'; ! *buf = '\0'; } return (saw_quote || buf > start_buf); --- 195,214 ---- *buf = '\0'; ! if (!saw_quote) { ! const char **entry; ! ! for (entry = keywords; *entry != NULL; entry++) ! { ! if (strcmp(start_buf, *entry) == 0) ! { ! /* append newline to a magical keyword */ ! *buf++ = '\n'; ! *buf = '\0'; ! break; ! } ! } } return (saw_quote || buf > start_buf); *************** *** 219,225 **** next_token(FILE *fp, char *buf, int bufsz, bool *initial_quote) * The result is a palloc'd string, or NULL if we have reached EOL. */ static char * ! next_token_expand(const char *filename, FILE *file) { char buf[MAX_TOKEN]; char *comma_str = pstrdup(""); --- 222,228 ---- * The result is a palloc'd string, or NULL if we have reached EOL. */ static char * ! next_token_expand(const char *filename, FILE *file, const char **keywords) { char buf[MAX_TOKEN]; char *comma_str = pstrdup(""); *************** *** 231,237 **** next_token_expand(const char *filename, FILE *file) do { ! if (!next_token(file, buf, sizeof(buf), &initial_quote)) break; got_something = true; --- 234,241 ---- do { ! if (!next_token(filename, file, buf, sizeof(buf), &initial_quote, ! keywords)) break; got_something = true; *************** *** 246,252 **** next_token_expand(const char *filename, FILE *file) /* Is this referencing a file? */ if (!initial_quote && buf[0] == '@' && buf[1] != '\0') ! incbuf = tokenize_inc_file(filename, buf + 1); else incbuf = pstrdup(buf); --- 250,256 ---- /* Is this referencing a file? */ if (!initial_quote && buf[0] == '@' && buf[1] != '\0') ! incbuf = tokenize_inc_file(filename, buf + 1, keywords); else incbuf = pstrdup(buf); *************** *** 273,279 **** next_token_expand(const char *filename, FILE *file) /* * Free memory used by lines/tokens (i.e., structure built by tokenize_file) */ ! static void free_lines(List **lines, List **line_nums) { /* --- 277,283 ---- /* * Free memory used by lines/tokens (i.e., structure built by tokenize_file) */ ! void free_lines(List **lines, List **line_nums) { /* *************** *** 318,324 **** free_lines(List **lines, List **line_nums) static char * tokenize_inc_file(const char *outer_filename, ! const char *inc_filename) { char *inc_fullname; FILE *inc_file; --- 322,328 ---- static char * tokenize_inc_file(const char *outer_filename, ! const char *inc_filename, const char **keywords) { char *inc_fullname; FILE *inc_file; *************** *** 348,354 **** tokenize_inc_file(const char *outer_filename, { ereport(LOG, (errcode_for_file_access(), ! errmsg("could not open secondary authentication file \"@%s\" as \"%s\": %m", inc_filename, inc_fullname))); pfree(inc_fullname); --- 352,358 ---- { ereport(LOG, (errcode_for_file_access(), ! errmsg("could not open secondary configuration file \"@%s\" as \"%s\": %m", inc_filename, inc_fullname))); pfree(inc_fullname); *************** *** 357,363 **** tokenize_inc_file(const char *outer_filename, } /* There is possible recursion here if the file contains @ */ ! tokenize_file(inc_fullname, inc_file, &inc_lines, &inc_line_nums); FreeFile(inc_file); pfree(inc_fullname); --- 361,368 ---- } /* There is possible recursion here if the file contains @ */ ! tokenize_file(inc_fullname, inc_file, &inc_lines, &inc_line_nums, ! keywords); FreeFile(inc_file); pfree(inc_fullname); *************** *** 404,412 **** tokenize_inc_file(const char *outer_filename, * * filename must be the absolute path to the target file. */ ! static void tokenize_file(const char *filename, FILE *file, ! List **lines, List **line_nums) { List *current_line = NIL; int line_number = 1; --- 409,417 ---- * * filename must be the absolute path to the target file. */ ! void tokenize_file(const char *filename, FILE *file, ! List **lines, List **line_nums, const char **keywords) { List *current_line = NIL; int line_number = 1; *************** *** 416,422 **** tokenize_file(const char *filename, FILE *file, while (!feof(file) && !ferror(file)) { ! buf = next_token_expand(filename, file); /* add token to list, unless we are at EOL or comment start */ if (buf) --- 421,427 ---- while (!feof(file) && !ferror(file)) { ! buf = next_token_expand(filename, file, keywords); /* add token to list, unless we are at EOL or comment start */ if (buf) *************** *** 1490,1496 **** load_hba(void) return false; } ! tokenize_file(HbaFileName, file, &hba_lines, &hba_line_nums); FreeFile(file); /* Now parse all the lines */ --- 1495,1501 ---- return false; } ! tokenize_file(HbaFileName, file, &hba_lines, &hba_line_nums, hba_keywords); FreeFile(file); /* Now parse all the lines */ *************** *** 1809,1815 **** load_ident(void) } else { ! tokenize_file(IdentFileName, file, &ident_lines, &ident_line_nums); FreeFile(file); } } --- 1814,1821 ---- } else { ! tokenize_file(IdentFileName, file, &ident_lines, &ident_line_nums, ! hba_keywords); FreeFile(file); } } *** a/src/backend/postmaster/postmaster.c --- b/src/backend/postmaster/postmaster.c *************** *** 1063,1069 **** PostmasterMain(int argc, char *argv[]) autovac_init(); /* ! * Load configuration files for client authentication. */ if (!load_hba()) { --- 1063,1069 ---- autovac_init(); /* ! * Load configuration files for client authentication and replication. */ if (!load_hba()) { *************** *** 1075,1080 **** PostmasterMain(int argc, char *argv[]) --- 1075,1085 ---- (errmsg("could not load pg_hba.conf"))); } load_ident(); + if (max_wal_senders > 0 && !load_standbys()) + { + ereport(FATAL, + (errmsg("could not load standbys.conf"))); + } /* * Remember postmaster startup time *************** *** 1713,1718 **** retry1: --- 1718,1725 ---- (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("invalid value for boolean option \"replication\""))); } + else if (strcmp(nameptr, "standby_name") == 0) + standby_name = pstrdup(valptr); else { /* Assume it's a generic GUC option */ *************** *** 2129,2134 **** SIGHUP_handler(SIGNAL_ARGS) --- 2136,2146 ---- load_ident(); + /* Reload standbys configuration file too */ + if (max_wal_senders > 0 && !load_standbys()) + ereport(WARNING, + (errmsg("standbys.conf not reloaded"))); + #ifdef EXEC_BACKEND /* Update the starting-point file for future children */ write_nondefault_variables(PGC_SIGHUP); *** a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c --- b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c *************** *** 47,55 **** static bool justconnected = false; static char *recvBuf = NULL; /* Prototypes for interface functions */ ! static bool libpqrcv_connect(char *conninfo, XLogRecPtr startpoint); static bool libpqrcv_receive(int timeout, unsigned char *type, char **buffer, int *len); static void libpqrcv_disconnect(void); /* Prototypes for private functions */ --- 47,57 ---- static char *recvBuf = NULL; /* Prototypes for interface functions */ ! static bool libpqrcv_connect(char *conninfo, XLogRecPtr startpoint, ! char *standbyName); static bool libpqrcv_receive(int timeout, unsigned char *type, char **buffer, int *len); + static void libpqrcv_send(const char *buffer, int nbytes); static void libpqrcv_disconnect(void); /* Prototypes for private functions */ *************** *** 64,73 **** _PG_init(void) { /* Tell walreceiver how to reach us */ if (walrcv_connect != NULL || walrcv_receive != NULL || ! walrcv_disconnect != NULL) elog(ERROR, "libpqwalreceiver already loaded"); walrcv_connect = libpqrcv_connect; walrcv_receive = libpqrcv_receive; walrcv_disconnect = libpqrcv_disconnect; } --- 66,76 ---- { /* Tell walreceiver how to reach us */ if (walrcv_connect != NULL || walrcv_receive != NULL || ! walrcv_send != NULL || walrcv_disconnect != NULL) elog(ERROR, "libpqwalreceiver already loaded"); walrcv_connect = libpqrcv_connect; walrcv_receive = libpqrcv_receive; + walrcv_send = libpqrcv_send; walrcv_disconnect = libpqrcv_disconnect; } *************** *** 75,98 **** _PG_init(void) * Establish the connection to the primary server for XLOG streaming */ static bool ! libpqrcv_connect(char *conninfo, XLogRecPtr startpoint) { ! char conninfo_repl[MAXCONNINFO + 37]; char *primary_sysid; char standby_sysid[32]; TimeLineID primary_tli; TimeLineID standby_tli; PGresult *res; char cmd[64]; /* ! * Connect using deliberately undocumented parameter: replication. The ! * database name is ignored by the server in replication mode, but specify ! * "replication" for .pgpass lookup. */ ! snprintf(conninfo_repl, sizeof(conninfo_repl), ! "%s dbname=replication replication=true", ! conninfo); streamConn = PQconnectdb(conninfo_repl); if (PQstatus(streamConn) != CONNECTION_OK) --- 78,107 ---- * Establish the connection to the primary server for XLOG streaming */ static bool ! libpqrcv_connect(char *conninfo, XLogRecPtr startpoint, char *standbyName) { ! char conninfo_repl[MAXCONNINFO + MAXSTANDBYNAME + 37]; char *primary_sysid; char standby_sysid[32]; TimeLineID primary_tli; TimeLineID standby_tli; + char *primary_rplMode; PGresult *res; char cmd[64]; /* ! * Connect using deliberately undocumented parameter: replication ! * and standby_name. The database name is ignored by the server in ! * replication mode, but specify "replication" for .pgpass lookup. */ ! if (standbyName) ! snprintf(conninfo_repl, sizeof(conninfo_repl), ! "%s dbname=replication replication=true standby_name=%s", ! conninfo, standbyName); ! else ! snprintf(conninfo_repl, sizeof(conninfo_repl), ! "%s dbname=replication replication=true", ! conninfo); streamConn = PQconnectdb(conninfo_repl); if (PQstatus(streamConn) != CONNECTION_OK) *************** *** 109,119 **** libpqrcv_connect(char *conninfo, XLogRecPtr startpoint) { PQclear(res); ereport(ERROR, ! (errmsg("could not receive database system identifier and timeline ID from " ! "the primary server: %s", PQerrorMessage(streamConn)))); } ! if (PQnfields(res) != 2 || PQntuples(res) != 1) { int ntuples = PQntuples(res); int nfields = PQnfields(res); --- 118,128 ---- { PQclear(res); ereport(ERROR, ! (errmsg("could not receive database system identifier, timeline ID and " ! "replication mode from the primary server: %s", PQerrorMessage(streamConn)))); } ! if (PQnfields(res) != 3 || PQntuples(res) != 1) { int ntuples = PQntuples(res); int nfields = PQnfields(res); *************** *** 121,131 **** libpqrcv_connect(char *conninfo, XLogRecPtr startpoint) PQclear(res); ereport(ERROR, (errmsg("invalid response from primary server"), ! errdetail("Expected 1 tuple with 2 fields, got %d tuples with %d fields.", ntuples, nfields))); } primary_sysid = PQgetvalue(res, 0, 0); primary_tli = pg_atoi(PQgetvalue(res, 0, 1), 4, 0); /* * Confirm that the system identifier of the primary is the same as ours. --- 130,141 ---- PQclear(res); ereport(ERROR, (errmsg("invalid response from primary server"), ! errdetail("Expected 1 tuple with 3 fields, got %d tuples with %d fields.", ntuples, nfields))); } primary_sysid = PQgetvalue(res, 0, 0); primary_tli = pg_atoi(PQgetvalue(res, 0, 1), 4, 0); + primary_rplMode = PQgetvalue(res, 0, 2); /* * Confirm that the system identifier of the primary is the same as ours. *************** *** 146,158 **** libpqrcv_connect(char *conninfo, XLogRecPtr startpoint) * recovery target timeline. */ standby_tli = GetRecoveryTargetTLI(); - PQclear(res); if (primary_tli != standby_tli) ereport(ERROR, (errmsg("timeline %u of the primary does not match recovery target timeline %u", primary_tli, standby_tli))); ThisTimeLineID = primary_tli; /* Start streaming from the point requested by startup process */ snprintf(cmd, sizeof(cmd), "START_REPLICATION %X/%X", startpoint.xlogid, startpoint.xrecoff); --- 156,180 ---- * recovery target timeline. */ standby_tli = GetRecoveryTargetTLI(); if (primary_tli != standby_tli) + { + PQclear(res); ereport(ERROR, (errmsg("timeline %u of the primary does not match recovery target timeline %u", primary_tli, standby_tli))); + } ThisTimeLineID = primary_tli; + /* + * Confirm that the passed replication mode is valid. + */ + rplMode = ReplicationModeNameGetValue(primary_rplMode); + PQclear(res); + if (rplMode == InvalidReplicationMode) + ereport(ERROR, + (errmsg("invalid replication mode \"%s\"", + primary_rplMode))); + /* Start streaming from the point requested by startup process */ snprintf(cmd, sizeof(cmd), "START_REPLICATION %X/%X", startpoint.xlogid, startpoint.xrecoff); *************** *** 398,400 **** libpqrcv_receive(int timeout, unsigned char *type, char **buffer, int *len) --- 420,437 ---- return true; } + + /* + * Send a message to XLOG stream. + * + * ereports on error. + */ + static void + libpqrcv_send(const char *buffer, int nbytes) + { + if (PQputCopyData(streamConn, buffer, nbytes) <= 0 || + PQflush(streamConn)) + ereport(ERROR, + (errmsg("could not send data to WAL stream: %s", + PQerrorMessage(streamConn)))); + } *** /dev/null --- b/src/backend/replication/standbys.conf.sample *************** *** 0 **** --- 1,35 ---- + # PostgreSQL Standbys Configuration File + # =================================================== + # + # Refer to the "Streaming Replication" section in the PostgreSQL + # documentation for a complete description of this file. A short + # synopsis follows. + # + # This file controls which replication mode each standby uses. + # Records are of the form: + # + # STANDBY-NAME REPLICATION-MODE + # + # (The uppercase items must be replaced by actual values.) + # + # STANDBY-NAME can be "all", standby name, or a comma-separated list + # thereof. + # + # REPLICATION-MODE specifies how long transaction commit waits for + # replication before the commit command returns a "success" to a + # client. The valid modes are "async", "recv", "fsync" and "replay". + # + # Standby name containing spaces, commas, quotes and other special + # characters must be quoted. Quoting one of the keyword "all" makes + # the name lose its special character, and just match standby with + # that name. + # + # This file is read on server startup and when the postmaster receives + # a SIGHUP signal. If you edit the file on a running system, you have + # to SIGHUP the postmaster for the changes to take effect. You can + # use "pg_ctl reload" to do that. + + # Put your actual configuration here + # ---------------------------------- + + # STANDBY-NAME REPLICATION-MODE *** a/src/backend/replication/walreceiver.c --- b/src/backend/replication/walreceiver.c *************** *** 57,62 **** bool am_walreceiver; --- 57,63 ---- /* libpqreceiver hooks to these when loaded */ walrcv_connect_type walrcv_connect = NULL; walrcv_receive_type walrcv_receive = NULL; + walrcv_send_type walrcv_send = NULL; walrcv_disconnect_type walrcv_disconnect = NULL; #define NAPTIME_PER_CYCLE 100 /* max sleep time between cycles (100ms) */ *************** *** 113,118 **** static void WalRcvDie(int code, Datum arg); --- 114,120 ---- static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len); static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr); static void XLogWalRcvFlush(void); + static void XLogWalRcvSendRecPtr(XLogRecPtr recptr); /* Signal handlers */ static void WalRcvSigHupHandler(SIGNAL_ARGS); *************** *** 158,164 **** void --- 160,168 ---- WalReceiverMain(void) { char conninfo[MAXCONNINFO]; + char standbyName[MAXSTANDBYNAME]; XLogRecPtr startpoint; + XLogRecPtr ackedpoint = {0, 0}; /* use volatile pointer to prevent code rearrangement */ volatile WalRcvData *walrcv = WalRcv; *************** *** 206,211 **** WalReceiverMain(void) --- 210,216 ---- /* Fetch information required to start streaming */ strlcpy(conninfo, (char *) walrcv->conninfo, MAXCONNINFO); + strlcpy(standbyName, (char *) walrcv->standbyName, MAXSTANDBYNAME); startpoint = walrcv->receivedUpto; SpinLockRelease(&walrcv->mutex); *************** *** 247,253 **** WalReceiverMain(void) /* Load the libpq-specific functions */ load_file("libpqwalreceiver", false); if (walrcv_connect == NULL || walrcv_receive == NULL || ! walrcv_disconnect == NULL) elog(ERROR, "libpqwalreceiver didn't initialize correctly"); /* --- 252,258 ---- /* Load the libpq-specific functions */ load_file("libpqwalreceiver", false); if (walrcv_connect == NULL || walrcv_receive == NULL || ! walrcv_send == NULL || walrcv_disconnect == NULL) elog(ERROR, "libpqwalreceiver didn't initialize correctly"); /* *************** *** 261,267 **** WalReceiverMain(void) /* Establish the connection to the primary for XLOG streaming */ EnableWalRcvImmediateExit(); ! walrcv_connect(conninfo, startpoint); DisableWalRcvImmediateExit(); /* Loop until end-of-streaming or error */ --- 266,272 ---- /* Establish the connection to the primary for XLOG streaming */ EnableWalRcvImmediateExit(); ! walrcv_connect(conninfo, startpoint, standbyName); DisableWalRcvImmediateExit(); /* Loop until end-of-streaming or error */ *************** *** 311,316 **** WalReceiverMain(void) --- 316,340 ---- */ XLogWalRcvFlush(); } + + /* + * If replication_mode is "replay", send the last WAL replay location + * to the primary, to acknowledge that replication has been completed + * up to that. This occurs only when WAL records were replayed since + * the last acknowledgement. + */ + if (rplMode == REPLICATION_MODE_REPLAY && + XLByteLT(ackedpoint, LogstreamResult.Flush)) + { + XLogRecPtr recptr; + + recptr = GetReplayRecPtr(); + if (XLByteLT(ackedpoint, recptr)) + { + XLogWalRcvSendRecPtr(recptr); + ackedpoint = recptr; + } + } } } *************** *** 406,411 **** XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len) --- 430,448 ---- buf += sizeof(WalDataMessageHeader); len -= sizeof(WalDataMessageHeader); + /* + * If replication_mode is "recv", send the last WAL receive + * location to the primary, to acknowledge that replication + * has been completed up to that. + */ + if (rplMode == REPLICATION_MODE_RECV) + { + XLogRecPtr endptr = msghdr.dataStart; + + XLByteAdvance(endptr, len); + XLogWalRcvSendRecPtr(endptr); + } + XLogWalRcvWrite(buf, len, msghdr.dataStart); break; } *************** *** 523,528 **** XLogWalRcvFlush(void) --- 560,573 ---- LogstreamResult.Flush = LogstreamResult.Write; + /* + * If replication_mode is "fsync", send the last WAL flush + * location to the primary, to acknowledge that replication + * has been completed up to that. + */ + if (rplMode == REPLICATION_MODE_FSYNC) + XLogWalRcvSendRecPtr(LogstreamResult.Flush); + /* Update shared-memory status */ SpinLockAcquire(&walrcv->mutex); walrcv->latestChunkStart = walrcv->receivedUpto; *************** *** 541,543 **** XLogWalRcvFlush(void) --- 586,609 ---- } } } + + /* Send the lsn to the primary server */ + static void + XLogWalRcvSendRecPtr(XLogRecPtr recptr) + { + static char *msgbuf = NULL; + WalAckMessageData msgdata; + + /* + * Allocate buffer that will be used for each output message if first + * time through. We do this just once to reduce palloc overhead. + * The buffer must be made large enough for maximum-sized messages. + */ + if (msgbuf == NULL) + msgbuf = palloc(1 + sizeof(WalAckMessageData)); + + msgbuf[0] = 'l'; + msgdata.ackEnd = recptr; + memcpy(msgbuf + 1, &msgdata, sizeof(WalAckMessageData)); + walrcv_send(msgbuf, 1 + sizeof(WalAckMessageData)); + } *** a/src/backend/replication/walreceiverfuncs.c --- b/src/backend/replication/walreceiverfuncs.c *************** *** 168,178 **** ShutdownWalRcv(void) /* * Request postmaster to start walreceiver. * ! * recptr indicates the position where streaming should begin, and conninfo ! * is a libpq connection string to use. */ void ! RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo) { /* use volatile pointer to prevent code rearrangement */ volatile WalRcvData *walrcv = WalRcv; --- 168,180 ---- /* * Request postmaster to start walreceiver. * ! * recptr indicates the position where streaming should begin, conninfo ! * is a libpq connection string to use, and standbyName is name of this ! * standby. */ void ! RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo, ! const char *standbyName) { /* use volatile pointer to prevent code rearrangement */ volatile WalRcvData *walrcv = WalRcv; *************** *** 196,201 **** RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo) --- 198,207 ---- strlcpy((char *) walrcv->conninfo, conninfo, MAXCONNINFO); else walrcv->conninfo[0] = '\0'; + if (standbyName != NULL) + strlcpy((char *) walrcv->standbyName, standbyName, MAXSTANDBYNAME); + else + walrcv->standbyName[0] = '\0'; walrcv->walRcvState = WALRCV_STARTING; walrcv->startTime = now; *** a/src/backend/replication/walsender.c --- b/src/backend/replication/walsender.c *************** *** 38,43 **** --- 38,44 ---- #include "access/xlog_internal.h" #include "catalog/pg_type.h" + #include "libpq/hba.h" #include "libpq/libpq.h" #include "libpq/pqformat.h" #include "libpq/pqsignal.h" *************** *** 61,66 **** static WalSnd *MyWalSnd = NULL; --- 62,68 ---- /* Global state */ bool am_walsender = false; /* Am I a walsender process ? */ + char *standby_name = NULL; /* Name of connected standby */ /* User-settable parameters for walsender */ int max_wal_senders = 0; /* the maximum number of concurrent walsenders */ *************** *** 84,94 **** static uint32 sendOff = 0; --- 86,114 ---- */ static XLogRecPtr sentPtr = {0, 0}; + /* + * How far have we completed replication already? This is also + * advertised in MyWalSnd->ackdPtr. This is not used in asynchronous + * replication case. + */ + static XLogRecPtr ackdPtr = {0, 0}; + /* Flags set by signal handlers for later service in main loop */ static volatile sig_atomic_t got_SIGHUP = false; static volatile sig_atomic_t shutdown_requested = false; static volatile sig_atomic_t ready_to_stop = false; + /* + * pre-parsed content of standbys configuration file: list of + * StandbysLine structs + */ + static List *parsed_standbys_lines = NIL; + + static const char *standbys_keywords[] = {"all", NULL}; + + /* Path of standbys configuration file (relative to $PGDATA) */ + #define STANDBYS_FILE "standbys.conf" + /* Signal handlers */ static void WalSndSigHupHandler(SIGNAL_ARGS); static void WalSndShutdownHandler(SIGNAL_ARGS); *************** *** 102,108 **** static void WalSndHandshake(void); static void WalSndKill(int code, Datum arg); static void XLogRead(char *buf, XLogRecPtr recptr, Size nbytes); static bool XLogSend(char *msgbuf, bool *caughtup); ! static void CheckClosedConnection(void); /* Main entry point for walsender process */ --- 122,132 ---- static void WalSndKill(int code, Datum arg); static void XLogRead(char *buf, XLogRecPtr recptr, Size nbytes); static bool XLogSend(char *msgbuf, bool *caughtup); ! static void ProcessStreamMsgs(StringInfo inMsg); ! ! static bool parse_standbys_line(List *line, int line_num, StandbysLine *parsedline); ! static void free_standbys_record(StandbysLine *record); ! static void clean_standbys_list(List *lines); /* Main entry point for walsender process */ *************** *** 209,227 **** WalSndHandshake(void) StringInfoData buf; char sysid[32]; char tli[11]; /* ! * Reply with a result set with one row, two columns. ! * First col is system ID, and second is timeline ID */ snprintf(sysid, sizeof(sysid), UINT64_FORMAT, GetSystemIdentifier()); snprintf(tli, sizeof(tli), "%u", ThisTimeLineID); /* Send a RowDescription message */ pq_beginmessage(&buf, 'T'); ! pq_sendint(&buf, 2, 2); /* 2 fields */ /* first field */ pq_sendstring(&buf, "systemid"); /* col name */ --- 233,254 ---- StringInfoData buf; char sysid[32]; char tli[11]; + char mode[8]; /* ! * Reply with a result set with one row, three columns. ! * First col is system ID, second is timeline ID, and ! * third is replication mode. */ snprintf(sysid, sizeof(sysid), UINT64_FORMAT, GetSystemIdentifier()); snprintf(tli, sizeof(tli), "%u", ThisTimeLineID); + snprintf(mode, sizeof(mode), "%s", ReplicationModeNames[rplMode]); /* Send a RowDescription message */ pq_beginmessage(&buf, 'T'); ! pq_sendint(&buf, 3, 2); /* 3 fields */ /* first field */ pq_sendstring(&buf, "systemid"); /* col name */ *************** *** 240,254 **** WalSndHandshake(void) pq_sendint(&buf, 4, 2); /* typlen */ pq_sendint(&buf, 0, 4); /* typmod */ pq_sendint(&buf, 0, 2); /* format code */ pq_endmessage(&buf); /* Send a DataRow message */ pq_beginmessage(&buf, 'D'); ! pq_sendint(&buf, 2, 2); /* # of columns */ pq_sendint(&buf, strlen(sysid), 4); /* col1 len */ pq_sendbytes(&buf, (char *) &sysid, strlen(sysid)); pq_sendint(&buf, strlen(tli), 4); /* col2 len */ pq_sendbytes(&buf, (char *) tli, strlen(tli)); pq_endmessage(&buf); /* Send CommandComplete and ReadyForQuery messages */ --- 267,292 ---- pq_sendint(&buf, 4, 2); /* typlen */ pq_sendint(&buf, 0, 4); /* typmod */ pq_sendint(&buf, 0, 2); /* format code */ + + /* third field */ + pq_sendstring(&buf, "replication_mode"); /* col name */ + pq_sendint(&buf, 0, 4); /* table oid */ + pq_sendint(&buf, 0, 2); /* attnum */ + pq_sendint(&buf, TEXTOID, 4); /* type oid */ + pq_sendint(&buf, -1, 2); /* typlen */ + pq_sendint(&buf, 0, 4); /* typmod */ + pq_sendint(&buf, 0, 2); /* format code */ pq_endmessage(&buf); /* Send a DataRow message */ pq_beginmessage(&buf, 'D'); ! pq_sendint(&buf, 3, 2); /* # of columns */ pq_sendint(&buf, strlen(sysid), 4); /* col1 len */ pq_sendbytes(&buf, (char *) &sysid, strlen(sysid)); pq_sendint(&buf, strlen(tli), 4); /* col2 len */ pq_sendbytes(&buf, (char *) tli, strlen(tli)); + pq_sendint(&buf, strlen(mode), 4); /* col3 len */ + pq_sendbytes(&buf, (char *) &mode, strlen(mode)); pq_endmessage(&buf); /* Send CommandComplete and ReadyForQuery messages */ *************** *** 286,295 **** WalSndHandshake(void) pq_flush(); /* ! * Initialize position to the received one, then the * xlog records begin to be shipped from that position */ ! sentPtr = recptr; /* break out of the loop */ replication_started = true; --- 324,333 ---- pq_flush(); /* ! * Initialize positions to the received one, then the * xlog records begin to be shipped from that position */ ! sentPtr = ackdPtr = recptr; /* break out of the loop */ replication_started = true; *************** *** 323,375 **** WalSndHandshake(void) } /* ! * Check if the remote end has closed the connection. */ static void ! CheckClosedConnection(void) { ! unsigned char firstchar; ! int r; ! r = pq_getbyte_if_available(&firstchar); ! if (r < 0) ! { ! /* unexpected error or EOF */ ! ereport(COMMERROR, ! (errcode(ERRCODE_PROTOCOL_VIOLATION), ! errmsg("unexpected EOF on standby connection"))); ! proc_exit(0); ! } ! if (r == 0) { ! /* no data available without blocking */ ! return; ! } - /* Handle the very limited subset of commands expected in this phase */ - switch (firstchar) - { /* * 'X' means that the standby is closing down the socket. */ ! case 'X': ! proc_exit(0); ! default: ! ereport(FATAL, ! (errcode(ERRCODE_PROTOCOL_VIOLATION), ! errmsg("invalid standby closing message type %d", ! firstchar))); } } /* Main loop of walsender process */ static int WalSndLoop(void) { char *output_message; bool caughtup = false; /* * Allocate buffer that will be used for each output message. We do this * just once to reduce palloc overhead. The buffer must be made large --- 361,482 ---- } /* ! * Process messages received from the standby. ! * ! * ereports on error. */ static void ! ProcessStreamMsgs(StringInfo inMsg) { ! bool acked = false; ! /* Loop to process successive complete messages available */ ! for (;;) { ! unsigned char firstchar; ! int r; ! ! r = pq_getbyte_if_available(&firstchar); ! if (r < 0) ! { ! /* unexpected error or EOF */ ! ereport(COMMERROR, ! (errcode(ERRCODE_PROTOCOL_VIOLATION), ! errmsg("unexpected EOF on standby connection"))); ! proc_exit(0); ! } ! if (r == 0) ! { ! /* no data available without blocking */ ! break; ! } ! ! /* Handle the very limited subset of commands expected in this phase */ ! switch (firstchar) ! { ! case 'd': /* CopyData message */ ! { ! unsigned char rpltype; ! ! /* ! * Read the message contents. This is expected to be done without ! * blocking because we've been able to get message type code. ! */ ! if (pq_getmessage(inMsg, 0)) ! proc_exit(0); /* suitable message already logged */ ! ! /* Read the replication message type from CopyData message */ ! rpltype = pq_getmsgbyte(inMsg); ! switch (rpltype) ! { ! case 'l': ! { ! WalAckMessageData *msgdata; ! ! msgdata = (WalAckMessageData *) pq_getmsgbytes(inMsg, sizeof(WalAckMessageData)); ! ! /* ! * Update local status. ! * ! * The ackd ptr received from standby should not ! * go backwards. ! */ ! if (XLByteLE(ackdPtr, msgdata->ackEnd)) ! ackdPtr = msgdata->ackEnd; ! else ! ereport(FATAL, ! (errmsg("replication completion location went back from " ! "%X/%X to %X/%X", ! ackdPtr.xlogid, ackdPtr.xrecoff, ! msgdata->ackEnd.xlogid, msgdata->ackEnd.xrecoff))); ! ! acked = true; /* also need to update shared position */ ! break; ! } ! default: ! ereport(FATAL, ! (errcode(ERRCODE_PROTOCOL_VIOLATION), ! errmsg("invalid replication message type %d", ! rpltype))); ! } ! break; ! } /* * 'X' means that the standby is closing down the socket. */ ! case 'X': ! proc_exit(0); ! default: ! ereport(FATAL, ! (errcode(ERRCODE_PROTOCOL_VIOLATION), ! errmsg("invalid standby closing message type %d", ! firstchar))); ! } } + + if (acked) + { + /* use volatile pointer to prevent code rearrangement */ + volatile WalSnd *walsnd = MyWalSnd; + + SpinLockAcquire(&walsnd->mutex); + walsnd->ackdPtr = ackdPtr; + SpinLockRelease(&walsnd->mutex); + } } /* Main loop of walsender process */ static int WalSndLoop(void) { + StringInfoData input_message; char *output_message; bool caughtup = false; + initStringInfo(&input_message); + /* * Allocate buffer that will be used for each output message. We do this * just once to reduce palloc overhead. The buffer must be made large *************** *** 438,444 **** WalSndLoop(void) /* Sleep and check that the connection is still alive */ pg_usleep(remain > NAPTIME_PER_CYCLE ? NAPTIME_PER_CYCLE : remain); ! CheckClosedConnection(); remain -= NAPTIME_PER_CYCLE; } --- 545,551 ---- /* Sleep and check that the connection is still alive */ pg_usleep(remain > NAPTIME_PER_CYCLE ? NAPTIME_PER_CYCLE : remain); ! ProcessStreamMsgs(&input_message); remain -= NAPTIME_PER_CYCLE; } *************** *** 497,502 **** InitWalSnd(void) --- 604,611 ---- MyWalSnd = (WalSnd *) walsnd; walsnd->pid = MyProcPid; MemSet(&MyWalSnd->sentPtr, 0, sizeof(XLogRecPtr)); + MemSet(&MyWalSnd->ackdPtr, 0, sizeof(XLogRecPtr)); + walsnd->rplMode = rplMode; SpinLockRelease(&walsnd->mutex); break; } *************** *** 523,528 **** WalSndKill(int code, Datum arg) --- 632,638 ---- * for this. */ MyWalSnd->pid = 0; + MyWalSnd->rplMode = InvalidReplicationMode; /* WalSnd struct isn't mine anymore */ MyWalSnd = NULL; *************** *** 896,938 **** WalSndShmemInit(void) } /* ! * This isn't currently used for anything. Monitoring tools might be ! * interested in the future, and we'll need something like this in the ! * future for synchronous replication. */ ! #ifdef NOT_USED /* ! * Returns the oldest Send position among walsenders. Or InvalidXLogRecPtr ! * if none. */ ! XLogRecPtr ! GetOldestWALSendPointer(void) { ! XLogRecPtr oldest = {0, 0}; ! int i; ! bool found = false; ! for (i = 0; i < max_wal_senders; i++) { ! /* use volatile pointer to prevent code rearrangement */ ! volatile WalSnd *walsnd = &WalSndCtl->walsnds[i]; ! XLogRecPtr recptr; ! if (walsnd->pid == 0) ! continue; ! SpinLockAcquire(&walsnd->mutex); ! recptr = walsnd->sentPtr; ! SpinLockRelease(&walsnd->mutex); ! if (recptr.xlogid == 0 && recptr.xrecoff == 0) ! continue; ! if (!found || XLByteLT(recptr, oldest)) ! oldest = recptr; ! found = true; } ! return oldest; } ! #endif --- 1006,1253 ---- } /* ! * Ensure that all xlog records through the given position is ! * replicated to the standby servers. ! * ! * XXX: We should replace the poll loop in this function with a latch. ! */ ! void ! WaitXLogSend(XLogRecPtr record) ! { ! Assert(max_wal_senders > 0); ! ! /* ! * XXX: We should track the number of currently connected standbys ! * and skip waiting if it's zero. ! */ ! ! for (;;) ! { ! int i; ! bool ackd = true; ! ! for (i = 0; i < max_wal_senders; i++) ! { ! /* use volatile pointer to prevent code rearrangement */ ! volatile WalSnd *walsnd = &WalSndCtl->walsnds[i]; ! XLogRecPtr recptr; ! ! /* Don't wait for unconnected and asynchronous standbys */ ! if (walsnd->pid == 0 || walsnd->rplMode <= REPLICATION_MODE_ASYNC) ! continue; ! ! SpinLockAcquire(&walsnd->mutex); ! recptr = walsnd->ackdPtr; ! SpinLockRelease(&walsnd->mutex); ! ! if (XLByteLT(recptr, record)) ! { ! ackd = false; ! break; ! } ! } ! ! if (ackd) ! return; ! ! pg_usleep(100000L); /* 100ms */ ! } ! } ! ! ! /* ---------- ! * Routines to handle standbys configuration file ! * ---------- ! */ ! ! /* ! * Scan the (pre-parsed) standbys configuration file line by line, ! * looking for a match to the standby name passed from the standby. */ ! bool ! check_standbys(void) ! { ! ListCell *line; ! StandbysLine *standbys; ! ! foreach(line, parsed_standbys_lines) ! { ! char *tok; ! ! standbys = (StandbysLine *) lfirst(line); ! ! /* Check standby name */ ! for (tok = strtok(standbys->standbyName, MULTI_VALUE_SEP); ! tok != NULL; ! tok = strtok(NULL, MULTI_VALUE_SEP)) ! { ! if (strcmp(tok, "all\n") == 0 || ! (standby_name != NULL && ! strcmp(tok, standby_name) == 0)) ! { ! rplMode = standbys->rplMode; ! return true; ! } ! } ! } ! return false; ! } ! /* ! * Parse one line in the standbys configuration file and store ! * the result in a StandbysLine structure. */ ! static bool ! parse_standbys_line(List *line, int line_num, StandbysLine *parsedline) { ! char *token; ! ListCell *line_item; ! line_item = list_head(line); ! ! parsedline->linenumber = line_num; ! ! /* Get the standby name. */ ! parsedline->standbyName = pstrdup(lfirst(line_item)); ! ! /* Get the mode. */ ! line_item = lnext(line_item); ! if (!line_item) { ! ereport(LOG, ! (errcode(ERRCODE_CONFIG_FILE_ERROR), ! errmsg("end-of-line before mode specification"), ! errcontext("line %d of configuration file \"%s\"", ! line_num, STANDBYS_FILE))); ! return false; ! } ! token = lfirst(line_item); ! parsedline->rplMode = ReplicationModeNameGetValue(token); ! if (parsedline->rplMode == InvalidReplicationMode) ! { ! ereport(LOG, ! (errcode(ERRCODE_CONFIG_FILE_ERROR), ! errmsg("invalid replication mode \"%s\"", ! token), ! errcontext("line %d of configuration file \"%s\"", ! line_num, STANDBYS_FILE))); ! return false; ! } ! /* Ignore remaining tokens */ ! return true; ! } ! ! /* ! * Free an StandbysLine structure ! */ ! static void ! free_standbys_record(StandbysLine *record) ! { ! if (record->standbyName) ! pfree(record->standbyName); ! pfree(record); ! } ! ! /* ! * Free all records on the parsed Standbys list ! */ ! static void ! clean_standbys_list(List *lines) ! { ! ListCell *line; ! ! foreach(line, lines) ! { ! StandbysLine *parsed = (StandbysLine *) lfirst(line); ! if (parsed) ! free_standbys_record(parsed); } ! list_free(lines); } ! /* ! * Read the config file and create a List of StandbysLine records for the contents. ! * ! * The configuration is read into a temporary list, and if any parse error occurs ! * the old list is kept in place and false is returned. Only if the whole file ! * parses Ok is the list replaced, and the function returns true. ! */ ! bool ! load_standbys(void) ! { ! FILE *file; ! List *standbys_lines = NIL; ! List *standbys_line_nums = NIL; ! ListCell *line, ! *line_num; ! List *new_parsed_lines = NIL; ! bool ok = true; ! ! /* Ignore standbys.conf if replication is not enabled */ ! if (max_wal_senders <= 0) ! return true; ! ! file = AllocateFile(STANDBYS_FILE, "r"); ! if (file == NULL) ! { ! ereport(LOG, ! (errcode_for_file_access(), ! errmsg("could not open configuration file \"%s\": %m", ! STANDBYS_FILE))); ! ! /* ! * Caller will take care of making this a FATAL error in case this is ! * the initial startup. If it happens on reload, we just keep the old ! * version around. ! */ ! return false; ! } ! ! tokenize_file(STANDBYS_FILE, file, &standbys_lines, &standbys_line_nums, ! standbys_keywords); ! FreeFile(file); ! ! /* Now parse all the lines */ ! forboth(line, standbys_lines, line_num, standbys_line_nums) ! { ! StandbysLine *newline; ! ! newline = palloc0(sizeof(StandbysLine)); ! ! if (!parse_standbys_line(lfirst(line), lfirst_int(line_num), newline)) ! { ! /* Parse error in the file, so indicate there's a problem */ ! free_standbys_record(newline); ! ok = false; ! ! /* ! * Keep parsing the rest of the file so we can report errors on ! * more than the first row. Error has already been reported in the ! * parsing function, so no need to log it here. ! */ ! continue; ! } ! ! new_parsed_lines = lappend(new_parsed_lines, newline); ! } ! ! /* Free the temporary lists */ ! free_lines(&standbys_lines, &standbys_line_nums); ! ! if (!ok) ! { ! /* Parsing failed at one or more rows, so bail out */ ! clean_standbys_list(new_parsed_lines); ! return false; ! } ! ! /* Loaded new file successfully, replace the one we use */ ! clean_standbys_list(parsed_standbys_lines); ! parsed_standbys_lines = new_parsed_lines; ! ! return true; ! } *** a/src/backend/utils/init/postinit.c --- b/src/backend/utils/init/postinit.c *************** *** 661,666 **** InitPostgres(const char *in_dbname, Oid dboid, const char *username, --- 661,688 ---- ereport(FATAL, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("must be superuser to start walsender"))); + + /* + * In EXEC_BACKEND case, we didn't inherit the contents of standbys.conf + * etcetera from the postmaster, and have to load them ourselves. Note we + * are loading them into the startup transaction's memory context, not + * PostmasterContext, but that shouldn't matter. + * + * FIXME: [fork/exec] Ugh. Is there a way around this overhead? + */ + #ifdef EXEC_BACKEND + if (!load_standbys()) + { + ereport(FATAL, + (errmsg("could not load standbys.conf"))); + } + #endif + + if (!check_standbys()) + ereport(FATAL, + (errmsg("no standbys.conf entry for standby name \"%s\"", + standby_name))); + /* report this backend in the PgBackendStatus array */ pgstat_bestart(); /* close the transaction we started above */ *** a/src/include/access/xlog.h --- b/src/include/access/xlog.h *************** *** 189,194 **** typedef enum --- 189,229 ---- extern XLogRecPtr XactLastRecEnd; + /* + * Replication mode. This is used to identify how long transaction + * commit should wait for replication. + * + * REPLICATION_MODE_ASYNC doesn't make transaction commit wait for + * replication, i.e., asynchronous replication. + * + * REPLICATION_MODE_RECV makes transaction commit wait for XLOG + * records to be received on the standby. + * + * REPLICATION_MODE_FSYNC makes transaction commit wait for XLOG + * records to be received and fsync'd on the standby. + * + * REPLICATION_MODE_REPLAY makes transaction commit wait for XLOG + * records to be received, fsync'd and replayed on the standby. + */ + typedef enum ReplicationMode + { + InvalidReplicationMode = -1, + REPLICATION_MODE_ASYNC = 0, + REPLICATION_MODE_RECV, + REPLICATION_MODE_FSYNC, + REPLICATION_MODE_REPLAY + + /* + * NOTE: if you add a new mode, change MAXREPLICATIONMODE below + * and update the ReplicationModeNames array in xlog.c + */ + } ReplicationMode; + + #define MAXREPLICATIONMODE REPLICATION_MODE_REPLAY + + extern const char *ReplicationModeNames[]; + extern ReplicationMode rplMode; + /* these variables are GUC parameters related to XLOG */ extern int CheckPointSegments; extern int wal_keep_segments; *************** *** 298,307 **** extern void XLogPutNextOid(Oid nextOid); --- 333,345 ---- extern XLogRecPtr GetRedoRecPtr(void); extern XLogRecPtr GetInsertRecPtr(void); extern XLogRecPtr GetFlushRecPtr(void); + extern XLogRecPtr GetReplayRecPtr(void); extern void GetNextXidAndEpoch(TransactionId *xid, uint32 *epoch); extern TimeLineID GetRecoveryTargetTLI(void); extern void HandleStartupProcInterrupts(void); extern void StartupProcessMain(void); + extern ReplicationMode ReplicationModeNameGetValue(char *name); + #endif /* XLOG_H */ *** a/src/include/libpq/hba.h --- b/src/include/libpq/hba.h *************** *** 15,20 **** --- 15,24 ---- #include "libpq/pqcomm.h" + /* This is used to separate values in multi-valued column strings */ + #define MULTI_VALUE_SEP "\001" + + typedef enum UserAuth { uaReject, *************** *** 89,93 **** extern int check_usermap(const char *usermap_name, --- 93,100 ---- const char *pg_role, const char *auth_user, bool case_sensitive); extern bool pg_isblank(const char c); + extern void tokenize_file(const char *filename, FILE *file, + List **lines, List **line_nums, const char **keywords); + extern void free_lines(List **lines, List **line_nums); #endif /* HBA_H */ *** a/src/include/replication/walprotocol.h --- b/src/include/replication/walprotocol.h *************** *** 50,53 **** typedef struct --- 50,63 ---- */ #define MAX_SEND_SIZE (XLOG_BLCKSZ * 16) + /* + * Body for a WAL acknowledgment message (message type 'l'). This is wrapped + * within a CopyData message at the FE/BE protocol level. + */ + typedef struct + { + /* End of WAL replicated to the standby */ + XLogRecPtr ackEnd; + } WalAckMessageData; + #endif /* _WALPROTOCOL_H */ *** a/src/include/replication/walreceiver.h --- b/src/include/replication/walreceiver.h *************** *** 26,31 **** extern bool am_walreceiver; --- 26,38 ---- #define MAXCONNINFO 1024 /* + * MAXSTANDBYNAME: maximum size of standby name. + * + * XXX: Should this move to pg_config_manual.h? + */ + #define MAXSTANDBYNAME 64 + + /* * Values for WalRcv->walRcvState. */ typedef enum *************** *** 71,89 **** typedef struct */ char conninfo[MAXCONNINFO]; slock_t mutex; /* locks shared variables shown above */ } WalRcvData; extern WalRcvData *WalRcv; /* libpqwalreceiver hooks */ ! typedef bool (*walrcv_connect_type) (char *conninfo, XLogRecPtr startpoint); extern PGDLLIMPORT walrcv_connect_type walrcv_connect; typedef bool (*walrcv_receive_type) (int timeout, unsigned char *type, char **buffer, int *len); extern PGDLLIMPORT walrcv_receive_type walrcv_receive; typedef void (*walrcv_disconnect_type) (void); extern PGDLLIMPORT walrcv_disconnect_type walrcv_disconnect; --- 78,106 ---- */ char conninfo[MAXCONNINFO]; + /* + * standby name; is used for the master to determine replication mode + * from standbys configuration file. + */ + char standbyName[MAXSTANDBYNAME]; + slock_t mutex; /* locks shared variables shown above */ } WalRcvData; extern WalRcvData *WalRcv; /* libpqwalreceiver hooks */ ! typedef bool (*walrcv_connect_type) (char *conninfo, XLogRecPtr startpoint, ! char *standbyName); extern PGDLLIMPORT walrcv_connect_type walrcv_connect; typedef bool (*walrcv_receive_type) (int timeout, unsigned char *type, char **buffer, int *len); extern PGDLLIMPORT walrcv_receive_type walrcv_receive; + typedef void (*walrcv_send_type) (const char *buffer, int nbytes); + extern PGDLLIMPORT walrcv_send_type walrcv_send; + typedef void (*walrcv_disconnect_type) (void); extern PGDLLIMPORT walrcv_disconnect_type walrcv_disconnect; *************** *** 93,99 **** extern void WalRcvShmemInit(void); extern void ShutdownWalRcv(void); extern bool WalRcvInProgress(void); extern XLogRecPtr WaitNextXLogAvailable(XLogRecPtr recptr, bool *finished); ! extern void RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo); extern XLogRecPtr GetWalRcvWriteRecPtr(XLogRecPtr *latestChunkStart); #endif /* _WALRECEIVER_H */ --- 110,117 ---- extern void ShutdownWalRcv(void); extern bool WalRcvInProgress(void); extern XLogRecPtr WaitNextXLogAvailable(XLogRecPtr recptr, bool *finished); ! extern void RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo, ! const char *standbyName); extern XLogRecPtr GetWalRcvWriteRecPtr(XLogRecPtr *latestChunkStart); #endif /* _WALRECEIVER_H */ *** a/src/include/replication/walsender.h --- b/src/include/replication/walsender.h *************** *** 22,27 **** typedef struct WalSnd --- 22,30 ---- { pid_t pid; /* this walsender's process id, or 0 */ XLogRecPtr sentPtr; /* WAL has been sent up to this point */ + XLogRecPtr ackdPtr; /* WAL has been replicated up to this point */ + + ReplicationMode rplMode; /* replication mode */ slock_t mutex; /* locks shared variables shown above */ } WalSnd; *************** *** 36,49 **** extern WalSndCtlData *WalSndCtl; --- 39,65 ---- /* global state */ extern bool am_walsender; + extern char *standby_name; /* user-settable parameters */ extern int WalSndDelay; extern int max_wal_senders; + /* struct definition for standbys configuration file */ + typedef struct + { + int linenumber; + char *standbyName; + ReplicationMode rplMode; + } StandbysLine; + extern int WalSenderMain(void); extern void WalSndSignals(void); extern Size WalSndShmemSize(void); extern void WalSndShmemInit(void); + extern void WaitXLogSend(XLogRecPtr record); + + extern bool check_standbys(void); + extern bool load_standbys(void); #endif /* _WALSENDER_H */ *** a/src/interfaces/libpq/fe-connect.c --- b/src/interfaces/libpq/fe-connect.c *************** *** 254,259 **** static const PQconninfoOption PQconninfoOptions[] = { --- 254,262 ---- {"replication", NULL, NULL, NULL, "Replication", "D", 5}, + {"standby_name", NULL, NULL, NULL, + "Standby-Name", "D", 64}, + /* Terminating entry --- MUST BE LAST */ {NULL, NULL, NULL, NULL, NULL, NULL, 0} *************** *** 613,618 **** fillPGconn(PGconn *conn, PQconninfoOption *connOptions) --- 616,623 ---- #endif tmp = conninfo_getval(connOptions, "replication"); conn->replication = tmp ? strdup(tmp) : NULL; + tmp = conninfo_getval(connOptions, "standby_name"); + conn->standbyName = tmp ? strdup(tmp) : NULL; } /* *************** *** 2622,2627 **** freePGconn(PGconn *conn) --- 2627,2634 ---- free(conn->dbName); if (conn->replication) free(conn->replication); + if (conn->standbyName) + free(conn->standbyName); if (conn->pguser) free(conn->pguser); if (conn->pgpass) *** a/src/interfaces/libpq/fe-exec.c --- b/src/interfaces/libpq/fe-exec.c *************** *** 2002,2007 **** PQnotifies(PGconn *conn) --- 2002,2010 ---- /* * PQputCopyData - send some data to the backend during COPY IN * + * This function can be called by walreceiver even during COPY OUT + * to send a message to the master. + * * Returns 1 if successful, 0 if data could not be sent (only possible * in nonblock mode), or -1 if an error occurs. */ *************** *** 2010,2016 **** PQputCopyData(PGconn *conn, const char *buffer, int nbytes) { if (!conn) return -1; ! if (conn->asyncStatus != PGASYNC_COPY_IN) { printfPQExpBuffer(&conn->errorMessage, libpq_gettext("no COPY in progress\n")); --- 2013,2020 ---- { if (!conn) return -1; ! if (conn->asyncStatus != PGASYNC_COPY_IN && ! conn->asyncStatus != PGASYNC_COPY_OUT) { printfPQExpBuffer(&conn->errorMessage, libpq_gettext("no COPY in progress\n")); *** a/src/interfaces/libpq/fe-protocol3.c --- b/src/interfaces/libpq/fe-protocol3.c *************** *** 1911,1916 **** build_startup_packet(const PGconn *conn, char *packet, --- 1911,1918 ---- ADD_STARTUP_OPTION("database", conn->dbName); if (conn->replication && conn->replication[0]) ADD_STARTUP_OPTION("replication", conn->replication); + if (conn->standbyName && conn->standbyName[0]) + ADD_STARTUP_OPTION("standby_name", conn->standbyName); if (conn->pgoptions && conn->pgoptions[0]) ADD_STARTUP_OPTION("options", conn->pgoptions); if (conn->send_appname) *** a/src/interfaces/libpq/libpq-int.h --- b/src/interfaces/libpq/libpq-int.h *************** *** 297,302 **** struct pg_conn --- 297,303 ---- char *fbappname; /* fallback application name */ char *dbName; /* database name */ char *replication; /* connect as the replication standby? */ + char *standbyName; /* standby name */ char *pguser; /* Postgres username and password, if any */ char *pgpass; char *keepalives; /* use TCP keepalives? */