diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c index 325239d..5b1e030 100644 --- a/src/backend/replication/syncrep.c +++ b/src/backend/replication/syncrep.c @@ -57,11 +57,17 @@ #include "utils/builtins.h" #include "utils/ps_status.h" +#define DEBUG_REPLICATION 1 + /* User-settable parameters for sync rep */ char *SyncRepStandbyNames; +int synchronous_replication_method; +int synchronous_standby_num; #define SyncStandbysDefined() \ - (SyncRepStandbyNames != NULL && SyncRepStandbyNames[0] != '\0') + (SyncRepStandbyNames != NULL && \ + SyncRepStandbyNames[0] != '\0' && \ + synchronous_standby_num > 0) static bool announce_next_takeover = true; @@ -348,58 +354,341 @@ SyncRepInitConfig(void) } } +/* Is this wal sender considerable one? */ +bool +SyncRepActiveListedWalSender(int num) +{ + volatile WalSnd *walsnd = &WalSndCtl->walsnds[num]; + + /* Must be active */ + if (walsnd->pid == 0) + return false; + + /* Must be streaming */ + if (walsnd->state != WALSNDSTATE_STREAMING) + return false; + + /* Must be synchronous */ + if (walsnd->sync_standby_priority == 0) + return false; + + /* Must have a valid flush position */ + if (XLogRecPtrIsInvalid(walsnd->flush)) + return false; + + return true; +} + /* - * 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. + * Get both LSNs: write and flush, according to replication method. + * And confirm whether we have advanced to LSN or not. */ -WalSnd * -SyncRepGetSynchronousStandby(void) +bool +SyncRepSyncedLsnAdvancedTo(XLogRecPtr *write_pos, XLogRecPtr *flush_pos) { - WalSnd *result = NULL; - int result_priority = 0; - int i; + XLogRecPtr tmp_write_pos; + XLogRecPtr tmp_flush_pos; + bool ret = false; + + /* '1-priority' method */ + if (synchronous_replication_method == SYNC_REP_METHOD_1_PRIORITY) + ret = SyncRepGetSyncLsnsOnePriority(&tmp_write_pos, &tmp_flush_pos); + + /* 'priority' method */ + else if (synchronous_replication_method == SYNC_REP_METHOD_PRIORITY) + ret = SyncRepGetSyncLsnsPriority(&tmp_write_pos, &tmp_flush_pos); + +#ifdef DEBUG_REPLICATION + elog(NOTICE, "====> CONSIDER (%d) write(tmp,my) %X/%X <= %X/%X, flush(tmp,my) : %X/%X <= %X/%X", + MyWalSnd->pid, + (uint32) (tmp_write_pos >> 32), (uint32) tmp_write_pos, + (uint32) (MyWalSnd->write >> 32), (uint32) MyWalSnd->write, + (uint32) (tmp_flush_pos >> 32), (uint32) tmp_flush_pos, + (uint32) (MyWalSnd->flush >> 32), (uint32) MyWalSnd->flush); +#endif + + /* Have we advanced LSN? */ + if (ret && + MyWalSnd->write >= tmp_write_pos && MyWalSnd->flush >= tmp_flush_pos) + { + *write_pos = tmp_write_pos; + *flush_pos = tmp_flush_pos; + + return true; + } + + return false; +} + +/* + * Obtain three palloc'd arrays containing position of standbys currently + * considered as synchronous, and its length. + */ +int +SyncRepGetSynchronousStandbys(int *sync_standbys) +{ + int num_sync = 0; + + + /* '1-priority' method */ + if (synchronous_replication_method == SYNC_REP_METHOD_1_PRIORITY) + num_sync = SyncRepGetSynchronousStandbysOnePriority(sync_standbys); + + /* 'priority' method */ + if (synchronous_replication_method == SYNC_REP_METHOD_PRIORITY) + num_sync = SyncRepGetSynchronousStandbysPriority(sync_standbys); + + return num_sync; +} + +/* + * Obtain standby currently considered as synchronous using + * '1-priority' method. + */ +int +SyncRepGetSynchronousStandbysOnePriority(int *sync_standbys) +{ + int priority = 0; + int i; 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) + /* Is this wal sender considerable one? */ + if (!SyncRepActiveListedWalSender(i)) continue; - /* Must be streaming */ - if (walsnd->state != WALSNDSTATE_STREAMING) - continue; + /* Find lowest priority standby */ + if (priority == 0 || + priority > walsnd->sync_standby_priority) + { + priority = walsnd->sync_standby_priority; + sync_standbys[0] = i; + } + } - /* Must be synchronous */ - this_priority = walsnd->sync_standby_priority; - if (this_priority == 0) - continue; +#ifdef DEBUG_REPLICATION + { + char *sync_list = palloc(sizeof(char) * 20); + char *tmp = sync_list; + int num_sync = 1; + + for (i = 0; i < num_sync; i++) + { + sprintf(tmp, "%d,", sync_standbys[i]); + tmp += 2; + } - /* Must have a lower priority value than any previous ones */ - if (result != NULL && result_priority <= this_priority) + *(tmp-1) = '\0'; + elog(NOTICE, "(%d) SyncedStandbyList [%s]", MyProcPid, sync_list); + pfree(sync_list); + } +#endif + + /* Always return 1 */ + return 1; +} + +/* + * Obtain standby currently considered as synchronous using + * 'priority' method. + */ +int +SyncRepGetSynchronousStandbysPriority(int *sync_standbys) +{ + int priority = 0; + int num_sync = 0; + int i; + + for (i = 0; i < max_wal_senders; i++) + { + /* Use volatile pointer to prevent code rearrangement */ + volatile WalSnd *walsnd = &WalSndCtl->walsnds[i]; + int j; + + /* Is this wal sender considerable one? */ + if (!SyncRepActiveListedWalSender(i)) continue; - /* Must have a valid flush position */ - if (XLogRecPtrIsInvalid(walsnd->flush)) + if (num_sync == synchronous_standby_num) + { + int new_priority = 0; + + for (j = 0; j < num_sync; j++) + { + volatile WalSnd *walsndloc = &WalSndCtl->walsnds[sync_standbys[j]]; + + /* Found sync standby */ + if (walsndloc->sync_standby_priority == priority && + walsnd->sync_standby_priority < priority) + sync_standbys[j] = i; + + /* Update highest priority standby */ + if (new_priority < walsndloc->sync_standby_priority) + new_priority = walsndloc->sync_standby_priority; + } + + priority = new_priority; + } + else + { + sync_standbys[num_sync] = i; + num_sync++; + + /* Keep track highest priority standby */ + if (priority < walsnd->sync_standby_priority) + priority = walsnd->sync_standby_priority; + } + } + +#ifdef DEBUG_REPLICATION + { + char *sync_list = palloc(sizeof(char) * 20); + char *tmp = sync_list; + + for (i = 0; i < num_sync; i++) + { + sprintf(tmp, "%d,", sync_standbys[i]); + tmp += 2; + } + + *(tmp-1) = '\0'; + elog(NOTICE, "(%d) SyncedStandbyList [%s]", MyProcPid, sync_list); + pfree(sync_list); + } +#endif + + return num_sync; +} + +/* + * Obtain currently synced LSN: write and flush, + * using '1-prioirty' method. + */ +bool +SyncRepGetSyncLsnsOnePriority(XLogRecPtr *write_pos, XLogRecPtr *flush_pos) +{ + int *sync_standbys = NULL; + int num_sync; + volatile WalSnd *walsnd; + + sync_standbys = (int *) palloc(sizeof(int) * synchronous_standby_num); + num_sync = SyncRepGetSynchronousStandbysOnePriority(sync_standbys); + + if (num_sync < synchronous_standby_num) + { + pfree(sync_standbys); + return false; + } + + /* Synchronous standby is always one */ + walsnd = &WalSndCtl->walsnds[sync_standbys[0]]; + + SpinLockAcquire(&walsnd->mutex); + *write_pos = walsnd->write; + *flush_pos = walsnd->flush; + SpinLockRelease(&walsnd->mutex); + +#ifdef DEBUG_REPLICATION + elog(NOTICE, "-----> LOCATION : write %X/%X, flush %X/%X", + (uint32) (*write_pos >> 32), (uint32) *write_pos, + (uint32) (*flush_pos >> 32), (uint32) *flush_pos); +#endif + + /* Clean up*/ + pfree(sync_standbys); + + return true; +} + +/* + * Obtain currently synced LSN: write and flush, + * using 'prioirty' method. + */ +bool +SyncRepGetSyncLsnsPriority(XLogRecPtr *write_pos, XLogRecPtr *flush_pos) +{ + int *sync_standbys = NULL; + int num_sync; + int i; + XLogRecPtr tmp_write = InvalidXLogRecPtr; + XLogRecPtr tmp_flush = InvalidXLogRecPtr; + + sync_standbys = (int *) palloc(sizeof(int) * synchronous_standby_num); + num_sync = SyncRepGetSynchronousStandbysPriority(sync_standbys); + + /* Just return, if sync standby is not enough */ + if (num_sync < synchronous_standby_num) + { + pfree(sync_standbys); + return false; + } + + for (i = 0; i < num_sync; i++) + { + volatile WalSnd *walsndloc = &WalSndCtl->walsnds[sync_standbys[i]]; + + SpinLockAcquire(&walsndloc->mutex); + + /* Store first candidate */ + if (XLogRecPtrIsInvalid(tmp_write) && XLogRecPtrIsInvalid(tmp_flush)) + { + tmp_write = walsndloc->write; + tmp_flush = walsndloc->flush; + SpinLockRelease(&walsndloc->mutex); continue; + } - result = (WalSnd *) walsnd; - result_priority = this_priority; + /* Find lowest XLogRecPtr of both write and flush from sync_nodes */ + if (tmp_write > walsndloc->write && tmp_flush > walsndloc->flush) + { + tmp_write = walsndloc->write; + tmp_flush = walsndloc->flush; + } - /* - * 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; + SpinLockRelease(&walsndloc->mutex); + } + + *write_pos = tmp_write; + *flush_pos = tmp_flush; + +#ifdef DEBUG_REPLICATION + elog(NOTICE, "================"); + elog(NOTICE, "WRITE LOCAITON"); + for (i = 0; i< num_sync; i++) + { + volatile WalSnd *walsndloc = &WalSndCtl->walsnds[sync_standbys[i]]; + + elog(NOTICE, " [%d] : %X/%X, priority = %d", i, + (uint32) (walsndloc->write >> 32) , + (uint32) walsndloc->write, + walsndloc->sync_standby_priority + ); + } + elog(NOTICE, "FLUSH LOCAITON"); + for (i = 0; i< num_sync; i++) + { + volatile WalSnd *walsndloc = &WalSndCtl->walsnds[sync_standbys[i]]; + + elog(NOTICE, " [%d] : %X/%X, priority = %d", i, + (uint32) (walsndloc->flush >> 32) , + (uint32) walsndloc->flush, + walsndloc->sync_standby_priority + ); } + elog(NOTICE, "-----> LOCATION : write %X/%X, flush %X/%X", + (uint32) (*write_pos >> 32), (uint32) *write_pos, + (uint32) (*flush_pos >> 32), (uint32) *flush_pos); - return result; + elog(NOTICE, "================"); +#endif + + /* Clean up*/ + pfree(sync_standbys); + + return true; } /* @@ -413,9 +702,9 @@ void SyncRepReleaseWaiters(void) { volatile WalSndCtlData *walsndctl = WalSndCtl; - WalSnd *syncWalSnd; - int numwrite = 0; - int numflush = 0; + XLogRecPtr write_pos = InvalidXLogRecPtr; + XLogRecPtr flush_pos = InvalidXLogRecPtr; + int numwrite, numflush; /* * If this WALSender is serving a standby that is not on the list of @@ -428,38 +717,35 @@ 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) + /* Get currently synced LSNs according to replication method */ + if (!(SyncRepSyncedLsnAdvancedTo(&write_pos, &flush_pos))) { LWLockRelease(SyncRepLock); - announce_next_takeover = true; return; } + +#ifdef DEBUG_REPLICATION + elog(NOTICE, "==========> DECIDE (%d) write : %X/%X, flush : %X/%X #####", + MyWalSnd->pid, + (uint32) (write_pos >> 32), (uint32) write_pos, + (uint32) (flush_pos >> 32), (uint32) flush_pos); +#endif + /* * 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); } @@ -480,6 +766,21 @@ SyncRepReleaseWaiters(void) (errmsg("standby \"%s\" is now the synchronous standby with priority %u", application_name, MyWalSnd->sync_standby_priority))); } + + + /*-----------------------------------*/ + + /* + for (i = 0; i < num_sync; i++) + { + volatile WalSnd *walsndloc = &WalSndCtl->walsnds[sync_standbys[i]]; + if (walsndloc == MyWalSnd) + { + found = true; + break; + } + } + */ } /* @@ -498,6 +799,7 @@ SyncRepGetStandbyPriority(void) ListCell *l; int priority = 0; bool found = false; + int num = 0; /* * Since synchronous cascade replication is not allowed, we always set the @@ -506,6 +808,10 @@ SyncRepGetStandbyPriority(void) if (am_cascading_walsender) return 0; + /* If no synchronous standby allowed, no cake for this WAL sender */ + if (!SyncStandbysDefined()) + return 0; + /* Need a modifiable copy of string */ rawstring = pstrdup(SyncRepStandbyNames); @@ -521,8 +827,16 @@ SyncRepGetStandbyPriority(void) foreach(l, elemlist) { - char *standby_name = (char *) lfirst(l); + char *standby_name; + + if (num == 0 && + synchronous_replication_method != SYNC_REP_METHOD_1_PRIORITY) + { + num = lfirst_int(l); + continue; + } + standby_name = (char *) lfirst(l); priority++; if (pg_strcasecmp(standby_name, application_name) == 0 || @@ -536,6 +850,9 @@ SyncRepGetStandbyPriority(void) pfree(rawstring); list_free(elemlist); +#ifdef DEBUG_REPLICATION + elog(NOTICE, "(%d) Got priority %d", MyWalSnd->pid, priority); +#endif return (found ? priority : 0); } @@ -684,11 +1001,42 @@ SyncRepQueueIsOrderedByLSN(int mode) * =========================================================== */ +void +assign_synchronous_standby_names(const char *newval, void *extra) +{ + char *rawstring; + List *elemlist; + + if (newval == NULL || newval[0] == '\0') + return; + + + /* Need a modifiable copy of string */ + rawstring = pstrdup(newval); + + /* Parse string into list of identifiers */ + SplitIdentifierString(rawstring, ',', &elemlist); + + /* Store them into globl variables */ + if (synchronous_replication_method == SYNC_REP_METHOD_1_PRIORITY) + synchronous_standby_num = 1; + else + synchronous_standby_num = pg_atoi(lfirst(list_head(elemlist)), sizeof(int), 0); + +#ifdef DEBUG_REPLICATION + elog(NOTICE, "Variable s_s_num : %d, s_s_names : %s(%s)", + synchronous_standby_num, rawstring, newval); +#endif + + return; +} + bool check_synchronous_standby_names(char **newval, void **extra, GucSource source) { char *rawstring; List *elemlist; + int num = 0; /* Need a modifiable copy of string */ rawstring = pstrdup(*newval); @@ -711,6 +1059,23 @@ check_synchronous_standby_names(char **newval, void **extra, GucSource source) * yet correctly set. */ + /* + * Check whether the number of synchronous stadby is + * specified correctly. + */ + if (elemlist != NULL && + synchronous_replication_method != SYNC_REP_METHOD_1_PRIORITY) + { + num = pg_atoi(lfirst(list_head(elemlist)), sizeof(int), 0); + + if (num < 1 || num > (list_length(elemlist) - 1)) + { + pfree(rawstring); + list_free(elemlist); + return false; + } + } + pfree(rawstring); list_free(elemlist); diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 4a4643e..1543571 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -2746,9 +2746,11 @@ 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 */ if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo)) ereport(ERROR, @@ -2774,11 +2776,13 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS) MemoryContextSwitchTo(oldcontext); + sync_standbys = (int *) palloc(sizeof(int) * synchronous_standby_num); + /* - * Get the currently active synchronous standby. + * Get the currently active synchronous standbys. */ LWLockAcquire(SyncRepLock, LW_SHARED); - sync_standby = SyncRepGetSynchronousStandby(); + num_sync = SyncRepGetSynchronousStandbys(sync_standbys); LWLockRelease(SyncRepLock); for (i = 0; i < max_wal_senders; i++) @@ -2848,18 +2852,34 @@ 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++) + { + /* Found sync standby */ + if (i == sync_standbys[j]) + { + values[7] = CStringGetTextDatum("sync"); + found = true; + break; + } + } + if (!found) + values[7] = CStringGetTextDatum("potential"); + } } tuplestore_putvalues(tupstore, tupdesc, values, nulls); } /* clean up and return the tuplestore */ + pfree(sync_standbys); tuplestore_donestoring(tupstore); + return (Datum) 0; } diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index a185749..a619aff 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -310,6 +310,12 @@ static const struct config_enum_entry xmloption_options[] = { {NULL, 0, false} }; +static const struct config_enum_entry synchronous_replication_method_options[] = { + {"1-priority", SYNC_REP_METHOD_1_PRIORITY, false}, + {"priority", SYNC_REP_METHOD_PRIORITY, false}, + {NULL, 0, false} +}; + /* * Although only "on", "off", and "safe_encoding" are documented, we * accept all the likely variants of "on" and "off". @@ -3361,7 +3367,7 @@ static struct config_string ConfigureNamesString[] = }, &SyncRepStandbyNames, "", - check_synchronous_standby_names, NULL, NULL + check_synchronous_standby_names, assign_synchronous_standby_names, NULL }, { @@ -3672,6 +3678,16 @@ static struct config_enum ConfigureNamesEnum[] = NULL, NULL, NULL }, + { + {"synchronous_replication_method", PGC_SIGHUP, REPLICATION_MASTER, + gettext_noop("Method for multiple synchronous replication."), + NULL + }, + &synchronous_replication_method, + SYNC_REP_METHOD_1_PRIORITY, synchronous_replication_method_options, + NULL, NULL, NULL + }, + /* End-of-list marker */ { {NULL, 0, 0, NULL, NULL}, NULL, 0, NULL, NULL, NULL, NULL diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 029114f..b79b965 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -241,6 +241,8 @@ #synchronous_standby_names = '' # standby servers that provide sync rep # comma-separated list of application_name # from standby(s); '*' = all +#synchronous_standby_num = 0 # number of standby servers using sync rep +#synchronous_replication_method = '1-priority' # 1-priority, priority #vacuum_defer_cleanup_age = 0 # number of xacts by which cleanup is delayed # - Standby Servers - diff --git a/src/include/replication/syncrep.h b/src/include/replication/syncrep.h index 71e2857..e1fa7ea 100644 --- a/src/include/replication/syncrep.h +++ b/src/include/replication/syncrep.h @@ -31,8 +31,14 @@ #define SYNC_REP_WAITING 1 #define SYNC_REP_WAIT_COMPLETE 2 +/* SyncRepMethod */ +#define SYNC_REP_METHOD_1_PRIORITY 0 +#define SYNC_REP_METHOD_PRIORITY 1 + /* user-settable parameters for synchronous replication */ extern char *SyncRepStandbyNames; +extern int synchronous_replication_method; +extern int synchronous_standby_num; /* called by user backend */ extern void SyncRepWaitForLSN(XLogRecPtr XactCommitLSN); @@ -49,9 +55,22 @@ extern void SyncRepUpdateSyncStandbysDefined(void); /* forward declaration to avoid pulling in walsender_private.h */ struct WalSnd; -extern struct WalSnd *SyncRepGetSynchronousStandby(void); + +extern int SyncRepGetSynchronousStandbys(int *sync_standbys); +extern bool SyncRepSyncedLsnAdvancedTo(XLogRecPtr *write_pos, XLogRecPtr *flush_pos); +extern bool SyncRepActiveListedWalSender(int num); + +/* 'priority' method */ +extern int SyncRepGetSynchronousStandbysPriority(int *sync_standbys); +extern bool SyncRepGetSyncLsnsPriority(XLogRecPtr *write_pos, XLogRecPtr *flush_pos); + +/* '1-priority' method */ +extern int SyncRepGetSynchronousStandbysOnePriority(int *sync_standbys); +extern bool SyncRepGetSyncLsnsOnePriority(XLogRecPtr *write_pos, XLogRecPtr *flush_pos); extern bool check_synchronous_standby_names(char **newval, void **extra, GucSource source); +extern void assign_synchronous_standby_names(const char *newva, void *extra); + extern void assign_synchronous_commit(int newval, void *extra); #endif /* _SYNCREP_H */