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/postmaster/postmaster.c b/src/backend/postmaster/postmaster.c index b16fc28..453f3f8 100644 --- a/src/backend/postmaster/postmaster.c +++ b/src/backend/postmaster/postmaster.c @@ -114,6 +114,7 @@ #include "postmaster/postmaster.h" #include "postmaster/syslogger.h" #include "replication/walsender.h" +#include "replication/syncrep.h" #include "storage/fd.h" #include "storage/ipc.h" #include "storage/pg_shmem.h" 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..a663f9b --- /dev/null +++ b/src/backend/replication/syncgroup_gram.y @@ -0,0 +1,136 @@ +%{ +/*------------------------------------------------------------------------- + * + * 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(int 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 +{ + int32 val; + char *str; + SyncGroupNode *expr; +} + +%token NAME +%token INT +%token AST + +%type result sync_list sync_list_ast sync_element sync_element_ast + sync_node_group + +%start result + +%% +result: + sync_node_group { SyncRepStandbys = $1; } +; + +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_node_group: + sync_list { $$ = create_group_node(1, $1); } + | sync_list_ast { $$ = create_group_node(1, $1);} + | INT '[' sync_list ']' { $$ = create_group_node($1, $3);} + | INT '[' sync_list_ast ']' { $$ = create_group_node($1, $3);} +; + +sync_element: + NAME { $$ = 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(int 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 = 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..922db6a --- /dev/null +++ b/src/backend/replication/syncgroup_scanner.l @@ -0,0 +1,159 @@ +%{ +#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 *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. + */ +node_name [a-zA-Z0-9\!\#\$\%\&\'\(\)\-\=\^\~\|\\\{\}\;\:\+\/\.] +dquoted_name [a-zA-Z0-9\!\#\$\%\&\'\(\)\-\=\^\~\|\\\{\}\;\:\+\/\.\,\[\]\ ] + +/* 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.val = atoi(yytext); + return INT; + } +{node_name}+ { + yylval.str = strdup(yytext); + return NAME; + } +{asterisk} { + yylval.str = strdup(yytext); + return AST; + } +{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_internal("%s", message))); +} + +void +syncgroup_scanner_init(const char *str) +{ + Size slen = strlen(str); + char *scanbuf; + + /* + * 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..ba2ac08 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_group. 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_group 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)) - { - /* 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) + if (SyncStandbysDefined()) { - 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,37 @@ 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; - } + syncgroup_scanner_init(*newval); + parse_rc = syncgroup_yyparse(); - /* - * 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. - */ + if (parse_rc != 0) + { + GUC_check_errdetail("Invalid syntax"); + return false; + } - pfree(rawstring); - list_free(elemlist); + syncgroup_scanner_finish(); + + /* + * 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_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 +898,24 @@ assign_synchronous_commit(int newval, void *extra) break; } } + +void +assign_synchronous_standby_names(const char *newval, void *extra) +{ + int parse_rc; + + 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 1dba7d9..455dea1 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');