diff --git a/doc/src/sgml/ref/allfiles.sgml b/doc/src/sgml/ref/allfiles.sgml index 01acc2e..6792eb0 100644 --- a/doc/src/sgml/ref/allfiles.sgml +++ b/doc/src/sgml/ref/allfiles.sgml @@ -181,6 +181,7 @@ Complete list of usable sgml source files in this directory. + diff --git a/doc/src/sgml/ref/waitlsn.sgml b/doc/src/sgml/ref/waitlsn.sgml new file mode 100644 index 0000000..077e869 --- /dev/null +++ b/doc/src/sgml/ref/waitlsn.sgml @@ -0,0 +1,119 @@ + + + + + WAITLSN + + + + WAITLSN + 7 + SQL - Language Statements + + + + WAITLSN + wait until target LSN has been replayed + + + + +WAITLSN 'LSN' [ , delay ] +WAITLSN_INFINITE 'LSN' +WAITLSN_NO_WAIT 'LSN' + + + + + Description + + + The WAITLSN wait till target LSN will + be replayed with an optional delay (milliseconds by default + infinity) to be wait for LSN to replayed. + + + + WAITLSN provides a simple + interprocess LSN wait mechanism for a backends on slave + in master-slave replication scheme on PostgreSQL database. + + + + The WAITLSN_INFINITE wait till target LSN will + be replayed on slave . + + + + The WAITLSN_NO_WAIT will tell if target LSN was replayed without any wait. + + + + + Parameters + + + + LSN + + + Target log sequence number to be wait for. + + + + + delay + + + Time in miliseconds to waiting for LSN to be replayed. + + + + + + + + Notes + + + Delay time for waiting till LSN to be replayed must be integer. For + default it is infinity. Waiting can be interupped using Ctl+C, or + by Postmaster death. + + + + + Examples + + + Configure and execute a waitlsn from + psql: + + +WAITLSN '0/3F07A6B1', 10000; +NOTICE: LSN is not reached. Try to make bigger delay. +WAITLSN + +WAITLSN '0/3F07A611'; +WAITLSN + +WAITLSN '0/3F0FF791', 500000; +^CCancel request sent +NOTICE: LSN is not reached. Try to make bigger delay. +ERROR: canceling statement due to user request + + + + + + Compatibility + + + There is no WAITLSN statement in the SQL + standard. + + + diff --git a/doc/src/sgml/reference.sgml b/doc/src/sgml/reference.sgml index 9000b3a..0c5951a 100644 --- a/doc/src/sgml/reference.sgml +++ b/doc/src/sgml/reference.sgml @@ -209,6 +209,7 @@ &update; &vacuum; &values; + &waitlsn; diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index df4843f..9dfb59d 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -40,6 +40,7 @@ #include "catalog/pg_control.h" #include "catalog/pg_database.h" #include "commands/tablespace.h" +#include "commands/waitlsn.h" #include "miscadmin.h" #include "pgstat.h" #include "port/atomics.h" @@ -147,7 +148,6 @@ const struct config_enum_entry sync_method_options[] = { {NULL, 0, false} }; - /* * Although only "on", "off", and "always" are documented, * we accept all the likely variants of "on" and "off". @@ -7237,6 +7237,15 @@ StartupXLOG(void) break; } + /* + * After update lastReplayedEndRecPtr set Latches in SHMEM array + */ + if (XLogCtl->lastReplayedEndRecPtr >= GetMinWaitLSN()) + { + + WaitLSNSetLatch(XLogCtl->lastReplayedEndRecPtr); + } + /* Else, try to fetch the next WAL record */ record = ReadRecord(xlogreader, InvalidXLogRecPtr, LOG, false); } while (record != NULL); diff --git a/src/backend/commands/Makefile b/src/backend/commands/Makefile index 4a6c99e..0d10117 100644 --- a/src/backend/commands/Makefile +++ b/src/backend/commands/Makefile @@ -20,6 +20,6 @@ OBJS = amcmds.o aggregatecmds.o alter.o analyze.o async.o cluster.o comment.o \ policy.o portalcmds.o prepare.o proclang.o publicationcmds.o \ schemacmds.o seclabel.o sequence.o statscmds.o subscriptioncmds.o \ tablecmds.o tablespace.o trigger.o tsearchcmds.o typecmds.o user.o \ - vacuum.o vacuumlazy.o variable.o view.o + vacuum.o vacuumlazy.o variable.o view.o waitlsn.o include $(top_srcdir)/src/backend/common.mk diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c index bacc08e..7d6011f 100644 --- a/src/backend/commands/async.c +++ b/src/backend/commands/async.c @@ -139,7 +139,6 @@ #include "utils/ps_status.h" #include "utils/timestamp.h" - /* * Maximum size of a NOTIFY payload, including terminating NULL. This * must be kept small enough so that a notification message fits on one diff --git a/src/backend/commands/waitlsn.c b/src/backend/commands/waitlsn.c new file mode 100644 index 0000000..4f6cb7b --- /dev/null +++ b/src/backend/commands/waitlsn.c @@ -0,0 +1,273 @@ +/*------------------------------------------------------------------------- + * + * waitlsn.c + * WaitLSN statment: WAITLSN + * + * Portions Copyright (c) 1996-2017, PostgreSQL Global Development Group + * Portions Copyright (c) 2017, Regents of PostgresPro + * + * IDENTIFICATION + * src/backend/commands/waitlsn.c + * + *------------------------------------------------------------------------- + */ + +/* + * ------------------------------------------------------------------------- + * Wait for LSN been replayed on slave + *------------------------------------------------------------------------- + */ + +#include "postgres.h" +#include "fmgr.h" +#include "pgstat.h" +#include "utils/pg_lsn.h" +#include "storage/latch.h" +#include "miscadmin.h" +#include "storage/spin.h" +#include "storage/backendid.h" +#include "access/xact.h" +#include "storage/shmem.h" +#include "storage/ipc.h" +#include "utils/timestamp.h" +#include "storage/pmsignal.h" +#include "access/xlog.h" +#include "access/xlogdefs.h" +#include "commands/waitlsn.h" +#include "storage/proc.h" +#include "access/transam.h" +#include "funcapi.h" +#include "catalog/pg_type.h" +#include "utils/builtins.h" + +/* Latches Own-DisownLatch and AbortCаllBack */ +static uint32 WaitLSNShmemSize(void); +static void WLDisownLatchAbort(XactEvent event, void *arg); +static void WLOwnLatch(XLogRecPtr trg_lsn); +static void WLDisownLatch(void); + +void _PG_init(void); + +/* Shared memory structures */ +typedef struct +{ + int pid; + volatile slock_t slock; + Latch latch; + XLogRecPtr trg_lsn; +} BIDLatch; + +typedef struct +{ + char dummy; // УБРАТЬ + int backend_maxid; + XLogRecPtr min_lsn; + BIDLatch l_arr[FLEXIBLE_ARRAY_MEMBER]; +} GlobState; + +static volatile GlobState *state; +bool is_latch_owned = false; + +/* Take Latch for current backend at the begining of WAITLSN */ +static void +WLOwnLatch(XLogRecPtr trg_lsn) +{ + SpinLockAcquire(&state->l_arr[MyBackendId].slock); + OwnLatch(&state->l_arr[MyBackendId].latch); + is_latch_owned = true; + + if (state->backend_maxid < MyBackendId) + state->backend_maxid = MyBackendId; + + state->l_arr[MyBackendId].pid = MyProcPid; + state->l_arr[MyBackendId].trg_lsn = trg_lsn; + SpinLockRelease(&state->l_arr[MyBackendId].slock); + + if (trg_lsn < state->min_lsn) + state->min_lsn = trg_lsn; +} + +/* Release Latch for current backend at the end of WAITLSN */ +static void +WLDisownLatch(void) +{ + int i; + XLogRecPtr trg_lsn = state->l_arr[MyBackendId].trg_lsn; + + SpinLockAcquire(&state->l_arr[MyBackendId].slock); + DisownLatch(&state->l_arr[MyBackendId].latch); + is_latch_owned = false; + state->l_arr[MyBackendId].pid = 0; + state->l_arr[MyBackendId].trg_lsn = InvalidXLogRecPtr; + + /* Update state->min_lsn iff it is nessesary choosing next min_lsn */ + if (state->min_lsn == trg_lsn) + { + state->min_lsn = PG_UINT64_MAX; + for (i = 2; i <= state->backend_maxid; i++) + if (state->l_arr[i].trg_lsn != InvalidXLogRecPtr && + state->l_arr[i].trg_lsn < state->min_lsn) + state->min_lsn = state->l_arr[i].trg_lsn; + } + + if (state->backend_maxid == MyBackendId) + for (i = (MaxConnections+1); i >=2; i--) + if (state->l_arr[i].pid != 0) + { + state->backend_maxid = i; + break; + } + + SpinLockRelease(&state->l_arr[MyBackendId].slock); +} + +/* CallBack function on abort*/ +static void +WLDisownLatchAbort(XactEvent event, void *arg) +{ + if (is_latch_owned && (event == XACT_EVENT_PARALLEL_ABORT || + event == XACT_EVENT_ABORT)) + { + WLDisownLatch(); + } +} + +/* Module load callback */ +void +_PG_init(void) +{ + if (!IsUnderPostmaster) + RegisterXactCallback(WLDisownLatchAbort, NULL); +} + +/* Get size of shared memory to room GlobState */ +static uint32 +WaitLSNShmemSize(void) +{ + return offsetof(GlobState, l_arr) + sizeof(BIDLatch) * (MaxConnections+1); +} + +/* Init array of Latches in shared memory */ +void +WaitLSNShmemInit(void) +{ + bool found; + uint i; + + state = (GlobState *) ShmemInitStruct("pg_wait_lsn", + WaitLSNShmemSize(), + &found); + if (!found) + { + for (i = 0; i < (MaxConnections+1); i++) + { + state->l_arr[i].pid = 0; + state->l_arr[i].trg_lsn = InvalidXLogRecPtr; + SpinLockInit(&state->l_arr[i].slock); + InitSharedLatch(&state->l_arr[i].latch); + } + state->backend_maxid = 0; + state->min_lsn = PG_UINT64_MAX; + } +} + +/* Set all Latches in shared memory cause new LSN been replayed*/ +void +WaitLSNSetLatch(XLogRecPtr cur_lsn) +{ + uint i; + + for (i = 2; i <= state->backend_maxid; i++) + { + if (state->l_arr[i].trg_lsn != 0) + { + SpinLockAcquire(&state->l_arr[i].slock); + if (state->l_arr[i].trg_lsn <= cur_lsn) + SetLatch(&state->l_arr[i].latch); + SpinLockRelease(&state->l_arr[i].slock); + } + } +} + +/* Get minimal LSN that will be next */ +XLogRecPtr +GetMinWaitLSN(void) +{ + return state->min_lsn; +} + +/* + * On WAITLSN own latch and wait till LSN is replayed, Postmaster death, interruption + * or timeout. + */ +void +WaitLSNUtility(const char *lsn, const int delay, DestReceiver *dest) +{ + XLogRecPtr trg_lsn = DatumGetLSN(DirectFunctionCall1(pg_lsn_in, CStringGetDatum(lsn))); + XLogRecPtr cur_lsn; + int latch_events; + uint64_t tdelay = delay; + long secs; + int microsecs; + TimestampTz timer = GetCurrentTimestamp(); + TupOutputState *tstate; + TupleDesc tupdesc; + char *value = "false"; + + if (delay > 0) + latch_events = WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH; + else + latch_events = WL_LATCH_SET | WL_POSTMASTER_DEATH; + + WLOwnLatch(trg_lsn); + + for (;;) + { + cur_lsn = GetXLogReplayRecPtr(NULL); + + /* If LSN had been Replayed */ + if (trg_lsn <= cur_lsn) + break; + + /* If the postmaster dies, finish immediately */ + if (!PostmasterIsAlive()) + break; + + /* If Delay time is over */ + if (latch_events & WL_TIMEOUT) + { + if (TimestampDifferenceExceeds(timer,GetCurrentTimestamp(),delay)) + break; + TimestampDifference(timer,GetCurrentTimestamp(),&secs, µsecs); + tdelay = delay - (secs*1000 + microsecs/1000); + } + + MyPgXact->xmin = InvalidTransactionId; + WaitLatch(&state->l_arr[MyBackendId].latch, latch_events, tdelay, WAIT_EVENT_CLIENT_READ); + ResetLatch(&state->l_arr[MyBackendId].latch); + + /* CHECK_FOR_INTERRUPTS if they comes then disown latch current */ + if (InterruptPending) + { + WLDisownLatch(); + ProcessInterrupts(); + } + + } + + WLDisownLatch(); + + if (trg_lsn > cur_lsn) + elog(NOTICE,"LSN is not reached. Try to make bigger delay."); + else + value = "true"; + + /* need a tuple descriptor representing a single TEXT column */ + tupdesc = CreateTemplateTupleDesc(1, false); + TupleDescInitEntry(tupdesc, (AttrNumber) 1, "LSN reached", TEXTOID, -1, 0); + /* prepare for projection of tuples */ + tstate = begin_tup_output_tupdesc(dest, tupdesc); + /* Send it */ + do_text_output_oneline(tstate, value); + end_tup_output(tstate); +} diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index 7d0de99..89e1e1d 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -275,7 +275,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); SecLabelStmt SelectStmt TransactionStmt TruncateStmt UnlistenStmt UpdateStmt VacuumStmt VariableResetStmt VariableSetStmt VariableShowStmt - ViewStmt CheckPointStmt CreateConversionStmt + ViewStmt WaitLSNStmt CheckPointStmt CreateConversionStmt DeallocateStmt PrepareStmt ExecuteStmt DropOwnedStmt ReassignOwnedStmt AlterTSConfigurationStmt AlterTSDictionaryStmt @@ -322,7 +322,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); %type OptSchemaEltList %type TriggerForSpec TriggerForType -%type TriggerActionTime +%type TriggerActionTime WaitDelay %type TriggerEvents TriggerOneEvent %type TriggerFuncArg %type TriggerWhen @@ -682,7 +682,8 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); VACUUM VALID VALIDATE VALIDATOR VALUE_P VALUES VARCHAR VARIADIC VARYING VERBOSE VERSION_P VIEW VIEWS VOLATILE - WHEN WHERE WHITESPACE_P WINDOW WITH WITHIN WITHOUT WORK WRAPPER WRITE + WAITLSN WAITLSN_INFINITE WAITLSN_NO_WAIT WHEN WHERE WHITESPACE_P WINDOW + WITH WITHIN WITHOUT WORK WRAPPER WRITE XML_P XMLATTRIBUTES XMLCONCAT XMLELEMENT XMLEXISTS XMLFOREST XMLNAMESPACES XMLPARSE XMLPI XMLROOT XMLSERIALIZE XMLTABLE @@ -933,6 +934,7 @@ stmt : | VariableSetStmt | VariableShowStmt | ViewStmt + | WaitLSNStmt | /*EMPTY*/ { $$ = NULL; } ; @@ -13819,7 +13821,41 @@ frame_bound: } ; +/***************************************************************************** + * + * QUERY: + * WAITLSN can appear as a query-level command + * + * + *****************************************************************************/ +WaitLSNStmt: + WAITLSN Sconst WaitDelay + { + WaitLSNStmt *n = makeNode(WaitLSNStmt); + n->lsn = $2; + n->delay = $3; + $$ = (Node *)n; + } + | WAITLSN_INFINITE Sconst + { + WaitLSNStmt *n = makeNode(WaitLSNStmt); + n->lsn = $2; + n->delay = 0; + $$ = (Node *)n; + } + | WAITLSN_NO_WAIT Sconst + { + WaitLSNStmt *n = makeNode(WaitLSNStmt); + n->lsn = $2; + n->delay = 1; + $$ = (Node *)n; + } + ; +WaitDelay: + ',' Iconst { $$ = $2; } + | /*EMPTY*/ { $$ = 0; } + ; /* * Supporting nonterminals for expressions. */ @@ -14856,6 +14892,9 @@ unreserved_keyword: | VIEW | VIEWS | VOLATILE + | WAITLSN + | WAITLSN_INFINITE + | WAITLSN_NO_WAIT | WHITESPACE_P | WITHIN | WITHOUT diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c index 2d1ed14..932136f 100644 --- a/src/backend/storage/ipc/ipci.c +++ b/src/backend/storage/ipc/ipci.c @@ -22,6 +22,7 @@ #include "access/subtrans.h" #include "access/twophase.h" #include "commands/async.h" +#include "commands/waitlsn.h" #include "miscadmin.h" #include "pgstat.h" #include "postmaster/autovacuum.h" @@ -271,6 +272,11 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port) AsyncShmemInit(); BackendRandomShmemInit(); + /* + * Init array of Latches in SHMEM for WAITLSN + */ + WaitLSNShmemInit(); + #ifdef EXEC_BACKEND /* diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c index 775477c..61f0674 100644 --- a/src/backend/tcop/utility.c +++ b/src/backend/tcop/utility.c @@ -56,6 +56,7 @@ #include "commands/user.h" #include "commands/vacuum.h" #include "commands/view.h" +#include "commands/waitlsn.h" #include "miscadmin.h" #include "parser/parse_utilcmd.h" #include "postmaster/bgwriter.h" @@ -923,6 +924,20 @@ standard_ProcessUtility(PlannedStmt *pstmt, break; } + case T_WaitLSNStmt: + { + WaitLSNStmt *stmt = (WaitLSNStmt *) parsetree; + if (!RecoveryInProgress()) + { + ereport(ERROR,(errcode(ERRCODE_READ_ONLY_SQL_TRANSACTION), + errmsg("cannot execute %s not during recovery", + "WaitLSN"))); + } + else + WaitLSNUtility(stmt->lsn, stmt->delay, dest); + } + break; + default: /* All other statement types have event trigger support */ ProcessUtilitySlow(pstate, pstmt, queryString, @@ -2481,6 +2496,10 @@ CreateCommandTag(Node *parsetree) tag = "NOTIFY"; break; + case T_WaitLSNStmt: + tag = "WAITLSN"; + break; + case T_ListenStmt: tag = "LISTEN"; break; @@ -3100,6 +3119,10 @@ GetCommandLogLevel(Node *parsetree) lev = LOGSTMT_ALL; break; + case T_WaitLSNStmt: + lev = LOGSTMT_ALL; + break; + case T_ListenStmt: lev = LOGSTMT_ALL; break; diff --git a/src/include/commands/waitlsn.h b/src/include/commands/waitlsn.h new file mode 100644 index 0000000..49cf9e8 --- /dev/null +++ b/src/include/commands/waitlsn.h @@ -0,0 +1,22 @@ +/*------------------------------------------------------------------------- + * + * waitlsn.h + * WaitLSN notification: WAITLSN + * + * Portions Copyright (c) 1996-2016, PostgreSQL Global Development Group + * Portions Copyright (c) 2016, Regents of PostgresPRO + * + * src/include/commands/waitlsn.h + * + *------------------------------------------------------------------------- + */ +#ifndef WAITLSN_H +#define WAITLSN_H +#include "tcop/dest.h" + +extern void WaitLSNUtility(const char *lsn, const int delay, DestReceiver *dest); +extern void WaitLSNShmemInit(void); +extern void WaitLSNSetLatch(XLogRecPtr cur_lsn); +extern XLogRecPtr GetMinWaitLSN(void); + +#endif /* WAITLSN_H */ diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h index 27bd4f3..df5d296 100644 --- a/src/include/nodes/nodes.h +++ b/src/include/nodes/nodes.h @@ -479,6 +479,7 @@ typedef enum NodeTag T_StartReplicationCmd, T_TimeLineHistoryCmd, T_SQLCmd, + T_WaitLSNStmt, /* * TAGS FOR RANDOM OTHER STUFF diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index 5f2a4a7..0ff537c 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -3424,4 +3424,16 @@ typedef struct DropSubscriptionStmt DropBehavior behavior; /* RESTRICT or CASCADE behavior */ } DropSubscriptionStmt; +/* ---------------------- + * WaitLSN Statement + * ---------------------- + */ +typedef struct WaitLSNStmt +{ + NodeTag type; + char *lsn; /* Taraget LSN to wait for */ + int delay; /* Delay to wait for LSN*/ + bool nowait; /* No wait for LSN just result*/ +} WaitLSNStmt; + #endif /* PARSENODES_H */ diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h index f50e45e..6ebb4a8 100644 --- a/src/include/parser/kwlist.h +++ b/src/include/parser/kwlist.h @@ -433,6 +433,9 @@ PG_KEYWORD("version", VERSION_P, UNRESERVED_KEYWORD) PG_KEYWORD("view", VIEW, UNRESERVED_KEYWORD) PG_KEYWORD("views", VIEWS, UNRESERVED_KEYWORD) PG_KEYWORD("volatile", VOLATILE, UNRESERVED_KEYWORD) +PG_KEYWORD("waitlsn", WAITLSN, UNRESERVED_KEYWORD) +PG_KEYWORD("waitlsn_infinite", WAITLSN_INFINITE, UNRESERVED_KEYWORD) +PG_KEYWORD("waitlsn_no_wait", WAITLSN_NO_WAIT, UNRESERVED_KEYWORD) PG_KEYWORD("when", WHEN, RESERVED_KEYWORD) PG_KEYWORD("where", WHERE, RESERVED_KEYWORD) PG_KEYWORD("whitespace", WHITESPACE_P, UNRESERVED_KEYWORD)