diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index a09ceb2..2197a94 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -2821,20 +2821,17 @@ include_dir 'conf.d' - Specifies a comma-separated list of standby names that can support - synchronous replication, as described in - . - At any one time there will be at most one active synchronous standby; - transactions waiting for commit will be allowed to proceed after - this standby server confirms receipt of their data. - The synchronous standby will be the first standby named in this list - that is both currently connected and streaming data in real-time - (as shown by a state of streaming in the + Specifies a standby names that can support synchronous replication using + either two types of syntax; comma-separated list or dedicated language, as + described in . + Transcations waiting for commit will be allowed to proceed after the + specified number of standby servers confirms receipt of their data. + The synchronous standbys will be those named in this parameter that are both + currently connected and streaming data in real-time (as shown by a state + of streaming in the pg_stat_replication view). - Other standby servers appearing later in this list represent potential - synchronous standbys. - If the current synchronous standby disconnects for whatever reason, + If the current any of synchronous standbys disconnects for whatever reason, it will be replaced immediately with the next-highest-priority standby. Specifying more than one standby name can allow very high availability. @@ -2842,9 +2839,10 @@ include_dir 'conf.d' The name of a standby server for this purpose is the application_name setting of the standby, as set in the primary_conninfo of the standby's WAL receiver. There is - no mechanism to enforce uniqueness. In case of duplicates one of the - matching standbys will be chosen to be the synchronous standby, though - exactly which one is indeterminate. + no mechanism to enforce uniqueness. For each specified standby name, + only the specified count of standbys will be chosen to be synchronous + standbys, though exactly which one is indeterminate, the rest will + represent potential synchronous standbys. The special entry * matches any application_name, including the default application name of walreceiver. diff --git a/doc/src/sgml/high-availability.sgml b/doc/src/sgml/high-availability.sgml index 6cb690c..216127d 100644 --- a/doc/src/sgml/high-availability.sgml +++ b/doc/src/sgml/high-availability.sgml @@ -1027,8 +1027,8 @@ primary_slot_name = 'node_a_slot' Synchronous replication offers the ability to confirm that all changes - made by a transaction have been transferred to one synchronous standby - server. This extends the standard level of durability + made by a transcation have been transferred to one or more synchronous standby + server. This extends that standard levelof durability offered by a transaction commit. This level of protection is referred to as 2-safe replication in computer science theory. @@ -1037,14 +1037,14 @@ primary_slot_name = 'node_a_slot' When requesting synchronous replication, each commit of a write transaction will wait until confirmation is received that the commit has been written to the transaction log on disk - of both the primary and standby server. The only possibility that data - can be lost is if both the primary and the standby suffer crashes at the + of both the primary and standby servers. The only possibility that data + can be lost is if both the primary and the standbys suffer crashes at the same time. This can provide a much higher level of durability, though only - if the sysadmin is cautious about the placement and management of the two + if the sysadmin is cautious about the placement and management of the these servers. Waiting for confirmation increases the user's confidence that the changes will not be lost in the event of server crashes but it also necessarily increases the response time for the requesting transaction. - The minimum wait time is the roundtrip time between primary to standby. + The minimum wait time is the roundtrip time between primary to standbys. @@ -1115,6 +1115,42 @@ primary_slot_name = 'node_a_slot' + + Multiple Synchronous Replication + + + Setting up synchronous standby at multiple locations bring us high availability. + It ensures that modified data will be replicated to multiple synchronous standbys. + + + + Multiple synchronous replication is set up by setting + using dedicated language. The syntax of dedicated language is following. + + + + synchronous_standby_names = 'N [ node [, ...] ]' + + + + Using dedicated language, we can define a synchronous group with a number N. + synchronous group can have some members which are consdiered as synchronous standby using comma-separated list. + Any standby name is accepted at any position of its list, but '*' is accepted at only tailing of the standby list. + The leading N is a number which specifies that how many standbys the master server waits to commit for. This number + must be less than actual number of members of its group. + The listed standby are given highest priority from left defined starting with 1. + + + + + All ASCII characters except for special characters(',', '"', '[', ']', ' ') are allowed as standby name. + When these special characters are used as standby name, whole standby name string need to be written in + double-quoted representation. + + + + + Planning for Performance diff --git a/src/backend/Makefile b/src/backend/Makefile index b3d5e2e..3e36686 100644 --- a/src/backend/Makefile +++ b/src/backend/Makefile @@ -203,7 +203,7 @@ distprep: $(MAKE) -C parser gram.c gram.h scan.c $(MAKE) -C bootstrap bootparse.c bootscanner.c $(MAKE) -C catalog schemapg.h postgres.bki postgres.description postgres.shdescription - $(MAKE) -C replication repl_gram.c repl_scanner.c + $(MAKE) -C replication syncgroup_gram.c syncgroup_scanner.c repl_gram.c repl_scanner.c $(MAKE) -C storage/lmgr lwlocknames.h $(MAKE) -C utils fmgrtab.c fmgroids.h errcodes.h $(MAKE) -C utils/misc guc-file.c @@ -319,6 +319,8 @@ maintainer-clean: distclean catalog/postgres.bki \ catalog/postgres.description \ catalog/postgres.shdescription \ + replication/syncgroup_gram.c \ + replication/syncgroup_scanner.c \ replication/repl_gram.c \ replication/repl_scanner.c \ storage/lmgr/lwlocknames.c \ diff --git a/src/backend/replication/.gitignore b/src/backend/replication/.gitignore index 2a0491d..00eb556 100644 --- a/src/backend/replication/.gitignore +++ b/src/backend/replication/.gitignore @@ -1,2 +1,4 @@ /repl_gram.c /repl_scanner.c +/syncgroup_gram.c +/syncgroup_scanner.c \ No newline at end of file diff --git a/src/backend/replication/Makefile b/src/backend/replication/Makefile index b73370e..ab0501b 100644 --- a/src/backend/replication/Makefile +++ b/src/backend/replication/Makefile @@ -15,7 +15,8 @@ include $(top_builddir)/src/Makefile.global override CPPFLAGS := -I. -I$(srcdir) $(CPPFLAGS) OBJS = walsender.o walreceiverfuncs.o walreceiver.o basebackup.o \ - repl_gram.o slot.o slotfuncs.o syncrep.o + syncgroup_gram.o repl_gram.o slot.o \ + slotfuncs.o syncrep.o SUBDIRS = logical @@ -23,6 +24,18 @@ include $(top_srcdir)/src/backend/common.mk # repl_scanner is compiled as part of repl_gram repl_gram.o: repl_scanner.c +# syncgroup_scanner is complied as part of syncgroup_gram +syncgroup_gram.o: syncgroup_scanner.c + +syncgroup_gram.h: syncgroup_gram.c ; +syncgroup_gram.c: BISONFLAGS += -d + +syncgroup_scanner.c: FLEXFLAGS = -CF -p -p +syncgroup_scanner.c: FLEX_NO_BACKUP=yes + +syncgroup_gram.o : syncgroup_gram.h # repl_gram.c and repl_scanner.c are in the distribution tarball, so # they are not cleaned here. +# syncgroup_gram.c and syncgroup_scanner.c are in the distribution tarball, so +# they are not cleaned here. diff --git a/src/backend/replication/syncgroup_gram.y b/src/backend/replication/syncgroup_gram.y new file mode 100644 index 0000000..470ecc6 --- /dev/null +++ b/src/backend/replication/syncgroup_gram.y @@ -0,0 +1,139 @@ +%{ +/*------------------------------------------------------------------------- + * + * syncgroup_gram.y - Parser for synchronous replication group + * + * Portions Copyright (c) 1996-2016, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * + * IDENTIFICATION + * src/backend/replication/syncgroup_gram.y + * + *------------------------------------------------------------------------- + */ + +#include "postgres.h" + +#include "replication/syncrep.h" + +static SyncGroupNode *create_name_node(char *name); +static SyncGroupNode *add_node(SyncGroupNode *node_list, SyncGroupNode *node); +static SyncGroupNode *create_group_node(char *wait_num, SyncGroupNode *node_list); + +/* + * Bison doesn't allocate anything that needs to live across parser calls, + * so we can easily have it use palloc instead of malloc. This prevents + * memory leaks if we error out during parsing. Note this only works with + * bison >= 2.0. However, in bison 1.875 the default is to use alloca() + * if possible, so there's not really much problem anyhow, at least if + * you're building with gcc. + */ +#define YYMALLOC palloc +#define YYFREE pfree + +%} + +%expect 0 +%name-prefix="syncgroup_yy" + +%union +{ + char *str; + SyncGroupNode *expr; +} + +%token NAME NUM +%token AST + +%type result sync_list sync_list_ast sync_element sync_element_ast + sync_node_group sync_group_old sync_group + +%start result + +%% +result: + sync_node_group { SyncRepStandbys = $1; } +; +sync_node_group: + sync_group_old { $$ = $1; } + | sync_group { $$ = $1; } +; +sync_group_old: + sync_list { $$ = create_group_node("1", $1); } + | sync_list_ast { $$ = create_group_node("1", $1); } +; +sync_group: + NUM '[' sync_list ']' { $$ = create_group_node($1, $3); } + | NUM '[' sync_list_ast ']' { $$ = create_group_node($1, $3); } +; +sync_list: + sync_element { $$ = $1;} + | sync_list ',' sync_element { $$ = add_node($1, $3);} +; +sync_list_ast: + sync_element_ast { $$ = $1;} + | sync_list ',' sync_element_ast { $$ = add_node($1, $3);} +; +sync_element: + NAME { $$ = create_name_node($1); } + | NUM { $$ = create_name_node($1); } +; +sync_element_ast: + AST { $$ = create_name_node($1); } +; +%% + +static SyncGroupNode * +create_name_node(char *name) +{ + SyncGroupNode *name_node = (SyncGroupNode *)malloc(sizeof(SyncGroupNode)); + + /* Common information */ + name_node->type = SYNC_REP_GROUP_NAME; + name_node->name = strdup(name); + name_node->next = NULL; + + /* For GROUP node */ + name_node->sync_method = 0; + name_node->wait_num = 0; + name_node->members = NULL; + name_node->SyncRepGetSyncedLsnsFn = NULL; + name_node->SyncRepGetSyncStandbysFn = NULL; + + return name_node; +} + +static SyncGroupNode * +create_group_node(char *wait_num, SyncGroupNode *node_list) +{ + SyncGroupNode *group_node = (SyncGroupNode *)malloc(sizeof(SyncGroupNode)); + + /* For NAME node */ + group_node->type = SYNC_REP_GROUP_GROUP | SYNC_REP_GROUP_MAIN; + group_node->name = "main"; + group_node->next = NULL; + + /* For GROUP node */ + group_node->sync_method = SYNC_REP_METHOD_PRIORITY; + group_node->wait_num = atoi(wait_num); + group_node->members = node_list; + group_node->SyncRepGetSyncedLsnsFn = SyncRepGetSyncedLsnsUsingPriority; + group_node->SyncRepGetSyncStandbysFn = SyncRepGetSyncStandbysUsingPriority; + + return group_node; +} + +static SyncGroupNode * +add_node(SyncGroupNode *node_list, SyncGroupNode *node) +{ + SyncGroupNode *tmp = node_list; + + /* Add node to tailing of node_list */ + while(tmp->next != NULL) tmp = tmp->next; + + tmp->next = node; + return node_list; +} + +#include "syncgroup_scanner.c" diff --git a/src/backend/replication/syncgroup_scanner.l b/src/backend/replication/syncgroup_scanner.l new file mode 100644 index 0000000..2d9f05f --- /dev/null +++ b/src/backend/replication/syncgroup_scanner.l @@ -0,0 +1,162 @@ +%{ +#include "postgres.h" + +/* No reason to constrain amount of data slurped */ +#define YY_READ_BUF_SIZE 16777216 + +#define BUFSIZE 8192 + +/* Handles to the buffer that the lexer uses internally */ +static YY_BUFFER_STATE scanbufhandle; + +/* Functions for handling double quoted string */ +static void init_xd_string(void); +static void addlit_xd_string(char *ytext, int yleng); +static void addlitchar_xd_string(unsigned char ychar); + +char *scanbuf; +char *xd_string; +int xd_size; /* actual size of xd_string */ +int xd_len; /* string length of xd_string */ +%} +%option 8bit +%option never-interactive +%option nounput +%option noinput +%option noyywrap +%option warn +%option prefix="syncgroup_yy" + +/* + * delimited identifiers (double-quoted identifiers) + */ +%x xd + +space [ \t\n\r\f] +non_newline [^\n\r] +whitespace ({space}+) +self [\[\]\,] +asterisk \* + +/* + * Basically all ascii characteres except for {self} and {whitespace} are allowed + * to be used for node name. These special charater could be used by double-quoted. + */ + /* excluding ' ', '\"', '*', ',', '[', ']' */ +node_name [\x21\x23-\x29\x29-\x2b\x2d-\x5a\x5c\x5e-\x7e] +/* excluding '\"' */ +dquoted_name [\x20\x21\x23-\x7e] + +/* Double-quoted string */ +dquote \" +xdstart {dquote} +xddouble {dquote}{dquote} +xdstop {dquote} +xdinside {dquoted_name}+ + +%% +{whitespace} { /* ignore */ } + +{xdstart} { + init_xd_string(); + BEGIN(xd); + } +{xddouble} { + addlitchar_xd_string('\"'); + } +{xdinside} { + addlit_xd_string(yytext, yyleng); + } +{xdstop} { + xd_string[xd_len] = '\0'; + yylval.str = xd_string; + BEGIN(INITIAL); + return NAME; + } +[1-9][0-9]* { + yylval.str = yytext; + return NUM; + } +{asterisk} { + yylval.str = strdup(yytext); + return AST; + } +{node_name}+ { + yylval.str = strdup(yytext); + return NAME; + } +{self} { + return yytext[0]; + } +. { + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("syntax error: unexpected character \"%s\"", yytext))); + } +%% + +void +yyerror(const char *message) +{ + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("%s at or near \"%s\" in \"%s\"", message, + yytext, scanbuf))); +} + +void +syncgroup_scanner_init(const char *str) +{ + Size slen = strlen(str); + + /* + * Might be left over after ereport() + */ + if (YY_CURRENT_BUFFER) + yy_delete_buffer(YY_CURRENT_BUFFER); + + /* + * Make a scan buffer with special termination needed by flex. + */ + scanbuf = (char *) palloc(slen + 2); + memcpy(scanbuf, str, slen); + scanbuf[slen] = scanbuf[slen + 1] = YY_END_OF_BUFFER_CHAR; + scanbufhandle = yy_scan_buffer(scanbuf, slen + 2); +} + +void +syncgroup_scanner_finish(void) +{ + yy_delete_buffer(scanbufhandle); + scanbufhandle = NULL; +} + +static void +init_xd_string() +{ + xd_string = palloc(sizeof(char) * BUFSIZE); + xd_size = BUFSIZE; + xd_len = 0; +} + +static void +addlit_xd_string(char *ytext, int yleng) +{ + /* enlarge buffer if needed */ + if ((xd_len + yleng) > xd_size) + xd_string = repalloc(xd_string, xd_size + BUFSIZE); + + memcpy(xd_string + xd_len, ytext, yleng); + xd_len += yleng; +} + +static void +addlitchar_xd_string(unsigned char ychar) +{ + /* enlarge buffer if needed */ + if ((xd_len + 1) > xd_size) + xd_string = repalloc(xd_string, xd_size + BUFSIZE); + + xd_string[xd_len] = ychar; + xd_len += 1; +} \ No newline at end of file diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c index 7f85b88..9086695 100644 --- a/src/backend/replication/syncrep.c +++ b/src/backend/replication/syncrep.c @@ -5,7 +5,7 @@ * Synchronous replication is new as of PostgreSQL 9.1. * * If requested, transaction commits wait until their commit LSN is - * acknowledged by the synchronous standby. + * acknowledged by the synchronous standbys. * * This module contains the code for waiting and release of backends. * All code in this module executes on the primary. The core streaming @@ -34,6 +34,13 @@ * synchronous standby it must have caught up with the primary; that may * take some time. Once caught up, the current highest priority standby * will release waiters from the queue. + * In 9.5 we support the possibility to have multiple synchronous standbys, + * as defined in synchronous_standby_names. Before on standby can become a + * synchronous standby it must have caught up with the primary; + * that may take some time. + * + * Waiters will be released from the queue once the number of standbys + * specified in synchronous_standby_names have caught. * * Portions Copyright (c) 2010-2016, PostgreSQL Global Development Group * @@ -58,10 +65,8 @@ #include "utils/ps_status.h" /* User-settable parameters for sync rep */ -char *SyncRepStandbyNames; - -#define SyncStandbysDefined() \ - (SyncRepStandbyNames != NULL && SyncRepStandbyNames[0] != '\0') +SyncGroupNode *SyncRepStandbys; +char *SyncRepStandbyNamesString; static bool announce_next_takeover = true; @@ -72,6 +77,10 @@ static void SyncRepCancelWait(void); static int SyncRepWakeQueue(bool all, int mode); static int SyncRepGetStandbyPriority(void); +static int SyncRepFindWalSenderByName(char *name); +static void SyncRepClearStandbyGroupList(SyncGroupNode *node); +static bool SyncRepSyncedLsnAdvancedTo(XLogRecPtr *write_pos, XLogRecPtr *flush_pos); +static bool SyncRepStandbyIsSync(int pos); #ifdef USE_ASSERT_CHECKING static bool SyncRepQueueIsOrderedByLSN(int mode); @@ -197,7 +206,7 @@ SyncRepWaitForLSN(XLogRecPtr XactCommitLSN) ereport(WARNING, (errcode(ERRCODE_ADMIN_SHUTDOWN), errmsg("canceling the wait for synchronous replication and terminating connection due to administrator command"), - errdetail("The transaction has already committed locally, but might not have been replicated to the standby."))); + errdetail("The transaction has already committed locally, but might not have been replicated to the standby(s)."))); whereToSendOutput = DestNone; SyncRepCancelWait(); break; @@ -214,7 +223,7 @@ SyncRepWaitForLSN(XLogRecPtr XactCommitLSN) QueryCancelPending = false; ereport(WARNING, (errmsg("canceling wait for synchronous replication due to user request"), - errdetail("The transaction has already committed locally, but might not have been replicated to the standby."))); + errdetail("The transaction has already committed locally, but might not have been replicated to the standby(s)."))); SyncRepCancelWait(); break; } @@ -318,6 +327,24 @@ SyncRepCleanupAtProcExit(void) } /* + * Clear all node in SyncRepStandbys recursively. + */ +static void +SyncRepClearStandbyGroupList(SyncGroupNode *group) +{ + SyncGroupNode *node = group->members; + + while (node != NULL) + { + SyncGroupNode *tmp = node->next; + + free(node); + node = tmp; + } +} + + +/* * =========================================================== * Synchronous Replication functions for wal sender processes * =========================================================== @@ -339,8 +366,11 @@ SyncRepInitConfig(void) priority = SyncRepGetStandbyPriority(); if (MyWalSnd->sync_standby_priority != priority) { + char *walsnd_name; LWLockAcquire(SyncRepLock, LW_EXCLUSIVE); MyWalSnd->sync_standby_priority = priority; + walsnd_name = (char *)MyWalSnd->name; + memcpy(walsnd_name, application_name, sizeof(MyWalSnd->name)); LWLockRelease(SyncRepLock); ereport(DEBUG1, (errmsg("standby \"%s\" now has synchronous standby priority %u", @@ -349,57 +379,58 @@ SyncRepInitConfig(void) } /* - * Find the WAL sender servicing the synchronous standby with the lowest - * priority value, or NULL if no synchronous standby is connected. If there - * are multiple standbys with the same lowest priority value, the first one - * found is selected. The caller must hold SyncRepLock. + * Check whether specified standby is active, which means not only having + * pid but also having any priority. */ -WalSnd * -SyncRepGetSynchronousStandby(void) +static bool +SyncRepStandbyIsSync(int pos) { - WalSnd *result = NULL; - int result_priority = 0; - int i; + volatile WalSnd *walsnd = &WalSndCtl->walsnds[pos]; - for (i = 0; i < max_wal_senders; i++) - { - /* Use volatile pointer to prevent code rearrangement */ - volatile WalSnd *walsnd = &WalSndCtl->walsnds[i]; - int this_priority; + /* Must be active */ + if (walsnd->pid == 0) + return false; - /* Must be active */ - if (walsnd->pid == 0) - continue; + /* Must be streaming */ + if (walsnd->state != WALSNDSTATE_STREAMING) + return false; - /* Must be streaming */ - if (walsnd->state != WALSNDSTATE_STREAMING) - continue; + /* Must be synchronous */ + if (walsnd->sync_standby_priority == 0) + return false; - /* Must be synchronous */ - this_priority = walsnd->sync_standby_priority; - if (this_priority == 0) - continue; + /* Must have a valid flush position */ + if (XLogRecPtrIsInvalid(walsnd->flush)) + return false; - /* Must have a lower priority value than any previous ones */ - if (result != NULL && result_priority <= this_priority) - continue; + return true; +} - /* Must have a valid flush position */ - if (XLogRecPtrIsInvalid(walsnd->flush)) - continue; +/* + * Finds the first active synchronous walsender with given name in + * WalSndCtl->walsnds and returns the index of that. Returns -1 if not found. + */ +static int +SyncRepFindWalSenderByName(char *name) +{ + int i; - result = (WalSnd *) walsnd; - result_priority = this_priority; + for (i = 0; i < max_wal_senders; i++) + { + /* Use volatile pointer to prevent code rearrangement */ + volatile WalSnd *walsnd = &WalSndCtl->walsnds[i]; + char *walsnd_name = (char *) walsnd->name; - /* - * If priority is equal to 1, there cannot be any other WAL senders - * with a lower priority, so we're done. - */ - if (this_priority == 1) - return result; + if (!SyncRepStandbyIsSync(i)) + continue; + + /* Compare wal sender name */ + if (pg_strcasecmp(name, walsnd_name) == 0) + return i; /* Found */ } - return result; + /* Not found */ + return -1; } /* @@ -413,7 +444,8 @@ void SyncRepReleaseWaiters(void) { volatile WalSndCtlData *walsndctl = WalSndCtl; - WalSnd *syncWalSnd; + XLogRecPtr write_pos = InvalidXLogRecPtr; + XLogRecPtr flush_pos = InvalidXLogRecPtr; int numwrite = 0; int numflush = 0; @@ -428,23 +460,11 @@ SyncRepReleaseWaiters(void) XLogRecPtrIsInvalid(MyWalSnd->flush)) return; - /* - * We're a potential sync standby. Release waiters if we are the highest - * priority standby. - */ LWLockAcquire(SyncRepLock, LW_EXCLUSIVE); - syncWalSnd = SyncRepGetSynchronousStandby(); - /* We should have found ourselves at least */ - Assert(syncWalSnd != NULL); - - /* - * If we aren't managing the highest priority standby then just leave. - */ - if (syncWalSnd != MyWalSnd) + if (!(SyncRepSyncedLsnAdvancedTo(&write_pos, &flush_pos))) { LWLockRelease(SyncRepLock); - announce_next_takeover = true; return; } @@ -452,14 +472,14 @@ SyncRepReleaseWaiters(void) * Set the lsn first so that when we wake backends they will release up to * this location. */ - if (walsndctl->lsn[SYNC_REP_WAIT_WRITE] < MyWalSnd->write) + if (walsndctl->lsn[SYNC_REP_WAIT_WRITE] < write_pos) { - walsndctl->lsn[SYNC_REP_WAIT_WRITE] = MyWalSnd->write; + walsndctl->lsn[SYNC_REP_WAIT_WRITE] = write_pos; numwrite = SyncRepWakeQueue(false, SYNC_REP_WAIT_WRITE); } - if (walsndctl->lsn[SYNC_REP_WAIT_FLUSH] < MyWalSnd->flush) + if (walsndctl->lsn[SYNC_REP_WAIT_FLUSH] < flush_pos) { - walsndctl->lsn[SYNC_REP_WAIT_FLUSH] = MyWalSnd->flush; + walsndctl->lsn[SYNC_REP_WAIT_FLUSH] = flush_pos; numflush = SyncRepWakeQueue(false, SYNC_REP_WAIT_FLUSH); } @@ -483,19 +503,172 @@ SyncRepReleaseWaiters(void) } /* + * Return true if we have enough synchrononized standbys and the 'safe' written + * flushed LSNs, which are LSNs assured in all standbys considered should be + * synchronized. + */ +static bool +SyncRepSyncedLsnAdvancedTo(XLogRecPtr *write_pos, XLogRecPtr *flush_pos) +{ + XLogRecPtr safe_write_pos; + XLogRecPtr safe_flush_pos; + bool got_lsns; + + /* Get synced LSNs at this moment */ + got_lsns = SyncRepStandbys->SyncRepGetSyncedLsnsFn(SyncRepStandbys, + &safe_write_pos, + &safe_flush_pos); + if (!got_lsns) + return false; + + /* Check whether each LSN has advanced to */ + if (MyWalSnd->write >= safe_write_pos) + *write_pos = safe_write_pos; + if (MyWalSnd->flush >= safe_flush_pos) + *flush_pos = safe_flush_pos; + + return true; +} + +/* + * Decide synced LSNs at this moment using priority method. + * If there are not active standbys enough to determine LSNs, return false. + */ +bool +SyncRepGetSyncedLsnsUsingPriority(SyncGroupNode *group, XLogRecPtr *write_pos, XLogRecPtr *flush_pos) +{ + int *sync_list = (int *)palloc(sizeof(int) * group->wait_num); + int sync_num; + int i; + + /* Get standbys list that are considered as synchronous at this moment */ + sync_num = group->SyncRepGetSyncStandbysFn(group, sync_list); + + /* If we could not get standbys enough, return false */ + if (sync_num != group->wait_num) + return false; + + *write_pos = InvalidXLogRecPtr; + *flush_pos = InvalidXLogRecPtr; + + /* + * In priority method, we seek the lowest each LSNs(write, flush) from + * standbys which are considered as synchronous. + */ + for (i = 0; i < sync_num; i++) + { + volatile WalSnd *walsnd = &WalSndCtl->walsnds[i]; + XLogRecPtr write; + XLogRecPtr flush; + + SpinLockAcquire(&walsnd->mutex); + write = walsnd->write; + flush = walsnd->flush; + SpinLockRelease(&walsnd->mutex); + + if (XLogRecPtrIsInvalid(*write_pos) || *write_pos > write) + *write_pos = write; + if (XLogRecPtrIsInvalid(*flush_pos) || *flush_pos > flush) + *flush_pos = flush; + } + + return true; +} + +/* + * Return the positions of the first group->wait_num synchronized standbys + * in group->member list into sync_list. sync_list is assumed to have enough + * space for at least group->wait_num elements. + */ +int +SyncRepGetSyncStandbysUsingPriority(SyncGroupNode *group, int *sync_list) +{ + SyncGroupNode *node; + int num = 0; + + for (node = group->members; node != NULL; node = node->next) + { + int pos = -1; + + /* We got enough synchronous standbys, return */ + if (num >= group->wait_num) + return num; + + if (pg_strcasecmp(node->name, "*") != 0) + { + pos = SyncRepFindWalSenderByName(node->name); + + /* Could not find wal sender by this name */ + if (pos == -1) + continue; + + sync_list[num] = pos; + num++; + } + else + { + /* + * '*' means that all remaining standbys up to group->wait_num + * are considered as synchronous. We allows user to use only '*' + * in synchronous_standby_names, or use '*' at last node of + * synchronous_standby_names. + */ + int i; + + for (i = 0; i < max_wal_senders; i++) + { + volatile WalSnd *walsnd = &WalSndCtl->walsnds[i]; + char *walsnd_name = (char *) walsnd->name; + bool already_listed = false; + int j; + + /* We got enough synchronous standbys, return */ + if (num >= group->wait_num) + return num; + + if (!SyncRepStandbyIsSync(i)) + continue; + + pos = SyncRepFindWalSenderByName(walsnd_name); + + /* Could not find wal sender by this name */ + if (pos == -1) + continue; + + for (j = 0; j < num; j++) + { + if (sync_list[j] == pos) + { + already_listed = true; + break; + } + } + + /* We already listed this standby, ignore */ + if (already_listed) + continue; + + /* Add this position to list */ + sync_list[num] = pos; + num++; + } + } + } + + return num; +} + +/* * Check if we are in the list of sync standbys, and if so, determine * priority sequence. Return priority if set, or zero to indicate that * we are not a potential sync standby. * - * Compare the parameter SyncRepStandbyNames against the application_name + * Compare the parameter SyncRepStandbys against the application_name * for this WALSender, or allow any name if we find a wildcard "*". */ static int SyncRepGetStandbyPriority(void) { - char *rawstring; - List *elemlist; - ListCell *l; int priority = 0; bool found = false; @@ -506,36 +679,23 @@ SyncRepGetStandbyPriority(void) if (am_cascading_walsender) return 0; - /* Need a modifiable copy of string */ - rawstring = pstrdup(SyncRepStandbyNames); - - /* Parse string into list of identifiers */ - if (!SplitIdentifierString(rawstring, ',', &elemlist)) + if (SyncStandbysDefined()) { - /* syntax error in list */ - pfree(rawstring); - list_free(elemlist); - /* GUC machinery will have already complained - no need to do again */ - return 0; - } - - foreach(l, elemlist) - { - char *standby_name = (char *) lfirst(l); - - priority++; + SyncGroupNode *node; - if (pg_strcasecmp(standby_name, application_name) == 0 || - pg_strcasecmp(standby_name, "*") == 0) + for (node = SyncRepStandbys->members; node != NULL; node = node->next) { - found = true; - break; + priority++; + + if (pg_strcasecmp(node->name, application_name) == 0 || + pg_strcasecmp(node->name, "*") == 0) + { + found = true; + break; + } } } - pfree(rawstring); - list_free(elemlist); - return (found ? priority : 0); } @@ -687,32 +847,55 @@ SyncRepQueueIsOrderedByLSN(int mode) bool check_synchronous_standby_names(char **newval, void **extra, GucSource source) { - char *rawstring; - List *elemlist; + int parse_rc; - /* Need a modifiable copy of string */ - rawstring = pstrdup(*newval); - - /* Parse string into list of identifiers */ - if (!SplitIdentifierString(rawstring, ',', &elemlist)) + if (*newval != NULL && (*newval)[0] != '\0') { - /* syntax error in list */ - GUC_check_errdetail("List syntax is invalid."); - pfree(rawstring); - list_free(elemlist); - return false; - } + SyncGroupNode *node; + int num_member = 0; - /* - * Any additional validation of standby names should go here. - * - * Don't attempt to set WALSender priority because this is executed by - * postmaster at startup, not WALSender, so the application_name is not - * yet correctly set. - */ + syncgroup_scanner_init(*newval); + parse_rc = syncgroup_yyparse(); + + if (parse_rc != 0) + { + GUC_check_errdetail("Invalid syntax"); + return false; + } + + syncgroup_scanner_finish(); - pfree(rawstring); - list_free(elemlist); + /* + * Any additional validation of standby names should go here. + * + * Don't attempt to set WALSender priority because this is executed by + * postmaster at startup, not WALSender, so the application_name is not + * yet correctly set. + */ + + /* + * Check whether group wait_num is not exceeded to the number of its + * member. + */ + for (node = SyncRepStandbys->members; node != NULL; node = node->next) + num_member++; + + if (num_member < SyncRepStandbys->wait_num) + { + SyncRepClearStandbyGroupList(SyncRepStandbys); + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + (errmsg_internal("The number of group memebers must be less than its group waits.")))); + } + + /* + * syncgroup_yyparse sets the global SyncRepStandbys as side effect. + * But this function is required to just check, so frees SyncRepStandbyNanes + * once parsing parameter. + */ + SyncRepClearStandbyGroupList(SyncRepStandbys); + SyncRepStandbys = NULL; + } return true; } @@ -733,3 +916,31 @@ assign_synchronous_commit(int newval, void *extra) break; } } + +void +assign_synchronous_standby_names(const char *newval, void *extra) +{ + int parse_rc; + + /* + * Before assign paramter, clear previous configuration, + * if there is. + */ + if (SyncRepStandbys) + SyncRepClearStandbyGroupList(SyncRepStandbys); + + if (newval != NULL && newval[0] != '\0') + { + syncgroup_scanner_init(newval); + parse_rc = syncgroup_yyparse(); + + if (parse_rc != 0) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + (errmsg_internal("Invalid syntax. synchronous_standby_names parse returned %d", + parse_rc)))); + + GUC_check_errdetail("Invalid syntax"); + syncgroup_scanner_finish(); + } +} diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index c03e045..58476d2 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -2735,7 +2735,6 @@ WalSndGetStateString(WalSndState state) return "UNKNOWN"; } - /* * Returns activity of walsenders, including pids and xlog locations sent to * standby servers. @@ -2749,7 +2748,8 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS) Tuplestorestate *tupstore; MemoryContext per_query_ctx; MemoryContext oldcontext; - WalSnd *sync_standby; + int *sync_standbys; + int num_sync; int i; /* check to see if caller supports us returning a tuplestore */ @@ -2780,9 +2780,13 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS) /* * Get the currently active synchronous standby. */ - LWLockAcquire(SyncRepLock, LW_SHARED); - sync_standby = SyncRepGetSynchronousStandby(); - LWLockRelease(SyncRepLock); + if (SyncStandbysDefined()) + { + sync_standbys = (int *) palloc(sizeof(int) * SyncRepStandbys->wait_num); + LWLockAcquire(SyncRepLock, LW_SHARED); + num_sync = SyncRepGetSyncStandbysUsingPriority(SyncRepStandbys, sync_standbys); + LWLockRelease(SyncRepLock); + } for (i = 0; i < max_wal_senders; i++) { @@ -2854,10 +2858,23 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS) */ if (priority == 0) values[7] = CStringGetTextDatum("async"); - else if (walsnd == sync_standby) - values[7] = CStringGetTextDatum("sync"); else - values[7] = CStringGetTextDatum("potential"); + { + int j; + bool found = false; + + for (j = 0; j < num_sync; j++) + { + if (sync_standbys[j] == i) + { + values[7] = CStringGetTextDatum("sync"); + found = true; + break; + } + } + if (!found) + values[7] = CStringGetTextDatum("potential"); + } } tuplestore_putvalues(tupstore, tupdesc, values, nulls); diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index ea5a09a..eaca66a 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -3384,9 +3384,9 @@ static struct config_string ConfigureNamesString[] = NULL, GUC_LIST_INPUT }, - &SyncRepStandbyNames, + &SyncRepStandbyNamesString, "", - check_synchronous_standby_names, NULL, NULL + check_synchronous_standby_names, assign_synchronous_standby_names, NULL }, { diff --git a/src/include/replication/syncrep.h b/src/include/replication/syncrep.h index 96e059b..74ae48c 100644 --- a/src/include/replication/syncrep.h +++ b/src/include/replication/syncrep.h @@ -31,8 +31,39 @@ #define SYNC_REP_WAITING 1 #define SYNC_REP_WAIT_COMPLETE 2 +/* SyncRepMethod */ +#define SYNC_REP_METHOD_PRIORITY 0 + +/* SyncGroupNode */ +#define SYNC_REP_GROUP_MAIN 0x01 +#define SYNC_REP_GROUP_NAME 0x02 +#define SYNC_REP_GROUP_GROUP 0x04 + +#define SyncStandbysDefined() \ + (SyncRepStandbyNamesString != NULL && SyncRepStandbyNamesString[0] != '\0') + +struct SyncGroupNode; +typedef struct SyncGroupNode SyncGroupNode; + +struct SyncGroupNode +{ + /* Common information */ + int type; + char *name; + SyncGroupNode *next; /* Same group, next name node */ + + /* For group ndoe */ + int sync_method; /* priority */ + int wait_num; + SyncGroupNode *members; /* member of its group */ + bool (*SyncRepGetSyncedLsnsFn) (SyncGroupNode *group, XLogRecPtr *write_pos, + XLogRecPtr *flush_pos); + int (*SyncRepGetSyncStandbysFn) (SyncGroupNode *group, int *list); +}; + /* user-settable parameters for synchronous replication */ -extern char *SyncRepStandbyNames; +extern SyncGroupNode *SyncRepStandbys; +extern char *SyncRepStandbyNamesString; /* called by user backend */ extern void SyncRepWaitForLSN(XLogRecPtr XactCommitLSN); @@ -47,11 +78,23 @@ extern void SyncRepReleaseWaiters(void); /* called by checkpointer */ extern void SyncRepUpdateSyncStandbysDefined(void); -/* forward declaration to avoid pulling in walsender_private.h */ -struct WalSnd; -extern struct WalSnd *SyncRepGetSynchronousStandby(void); - extern bool check_synchronous_standby_names(char **newval, void **extra, GucSource source); +extern void assign_synchronous_standby_names(const char *newval, void *extra); extern void assign_synchronous_commit(int newval, void *extra); +/* + * Internal functions for parsing the replication grammar, in syncgroup_gram.y and + * syncgroup_scanner.l + */ +extern int syncgroup_yyparse(void); +extern int syncgroup_yylex(void); +extern void syncgroup_yyerror(const char *str) pg_attribute_noreturn(); +extern void syncgroup_scanner_init(const char *query_string); +extern void syncgroup_scanner_finish(void); + +/* function for synchronous replication group */ +extern bool SyncRepGetSyncedLsnsUsingPriority(SyncGroupNode *group, + XLogRecPtr *write_pos, XLogRecPtr *flush_pos); +extern int SyncRepGetSyncStandbysUsingPriority(SyncGroupNode *group, int *sync_list); + #endif /* _SYNCREP_H */ diff --git a/src/include/replication/walsender_private.h b/src/include/replication/walsender_private.h index 7794aa5..97a71a8 100644 --- a/src/include/replication/walsender_private.h +++ b/src/include/replication/walsender_private.h @@ -19,6 +19,8 @@ #include "storage/shmem.h" #include "storage/spin.h" +#define MAX_WALSENDER_NAME 8192 + typedef enum WalSndState { WALSNDSTATE_STARTUP = 0, @@ -62,6 +64,11 @@ typedef struct WalSnd * SyncRepLock. */ int sync_standby_priority; + + /* + * Corresponding standby's application_name. + */ + const char name[NAMEDATALEN]; } WalSnd; extern WalSnd *MyWalSnd; diff --git a/src/tools/msvc/Mkvcbuild.pm b/src/tools/msvc/Mkvcbuild.pm index e4fb44e..a13bb18 100644 --- a/src/tools/msvc/Mkvcbuild.pm +++ b/src/tools/msvc/Mkvcbuild.pm @@ -149,6 +149,8 @@ sub mkvcbuild $postgres->AddFiles('src/backend/utils/misc', 'guc-file.l'); $postgres->AddFiles('src/backend/replication', 'repl_scanner.l', 'repl_gram.y'); + $postgres->AddFiles('src/backend/replication', 'syncgroup_scanner.l', + 'syncgroup_gram.y'); $postgres->AddDefine('BUILDING_DLL'); $postgres->AddLibrary('secur32.lib'); $postgres->AddLibrary('ws2_32.lib');