PostgreSQL Source Code git master
Loading...
Searching...
No Matches
Functions
wait.c File Reference
#include "postgres.h"
#include <math.h>
#include "access/xlog.h"
#include "access/xlogrecovery.h"
#include "access/xlogwait.h"
#include "catalog/pg_type_d.h"
#include "commands/defrem.h"
#include "commands/wait.h"
#include "executor/executor.h"
#include "parser/parse_node.h"
#include "storage/proc.h"
#include "utils/builtins.h"
#include "utils/guc.h"
#include "utils/pg_lsn.h"
#include "utils/snapmgr.h"
Include dependency graph for wait.c:

Go to the source code of this file.

Functions

void ExecWaitStmt (ParseState *pstate, WaitStmt *stmt, DestReceiver *dest)
 
TupleDesc WaitStmtResultDesc (WaitStmt *stmt)
 

Function Documentation

◆ ExecWaitStmt()

void ExecWaitStmt ( ParseState *  pstate,
WaitStmt *  stmt,
DestReceiver *  dest 
)

Definition at line 34 of file wait.c.

35{
36 XLogRecPtr lsn;
37 int64 timeout = 0;
38 WaitLSNResult waitLSNResult;
39 WaitLSNType lsnType = WAIT_LSN_TYPE_STANDBY_REPLAY; /* default */
40 bool throw = true;
41 TupleDesc tupdesc;
42 TupOutputState *tstate;
43 const char *result = "<unset>";
44 bool timeout_specified = false;
45 bool no_throw_specified = false;
46 bool mode_specified = false;
47
48 /* Parse and validate the mandatory LSN */
49 lsn = DatumGetLSN(DirectFunctionCall1(pg_lsn_in,
50 CStringGetDatum(stmt->lsn_literal)));
51
52 foreach_node(DefElem, defel, stmt->options)
53 {
54 if (strcmp(defel->defname, "mode") == 0)
55 {
56 char *mode_str;
57
58 if (mode_specified)
59 errorConflictingDefElem(defel, pstate);
60 mode_specified = true;
61
62 mode_str = defGetString(defel);
63
64 if (pg_strcasecmp(mode_str, "standby_replay") == 0)
65 lsnType = WAIT_LSN_TYPE_STANDBY_REPLAY;
66 else if (pg_strcasecmp(mode_str, "standby_write") == 0)
67 lsnType = WAIT_LSN_TYPE_STANDBY_WRITE;
68 else if (pg_strcasecmp(mode_str, "standby_flush") == 0)
69 lsnType = WAIT_LSN_TYPE_STANDBY_FLUSH;
70 else if (pg_strcasecmp(mode_str, "primary_flush") == 0)
71 lsnType = WAIT_LSN_TYPE_PRIMARY_FLUSH;
72 else
73 ereport(ERROR,
74 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
75 errmsg("unrecognized value for %s option \"%s\": \"%s\"",
76 "WAIT", defel->defname, mode_str),
77 parser_errposition(pstate, defel->location)));
78 }
79 else if (strcmp(defel->defname, "timeout") == 0)
80 {
81 char *timeout_str;
82 const char *hintmsg;
83 double result;
84
85 if (timeout_specified)
86 errorConflictingDefElem(defel, pstate);
87 timeout_specified = true;
88
89 timeout_str = defGetString(defel);
90
91 if (!parse_real(timeout_str, &result, GUC_UNIT_MS, &hintmsg))
92 {
93 ereport(ERROR,
94 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
95 errmsg("invalid timeout value: \"%s\"", timeout_str),
96 hintmsg ? errhint("%s", _(hintmsg)) : 0);
97 }
98
99 /*
100 * Get rid of any fractional part in the input. This is so we
101 * don't fail on just-out-of-range values that would round into
102 * range.
103 */
104 result = rint(result);
105
106 /* Range check */
107 if (unlikely(isnan(result) || !FLOAT8_FITS_IN_INT64(result)))
108 ereport(ERROR,
109 errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
110 errmsg("timeout value is out of range"));
111
112 if (result < 0)
113 ereport(ERROR,
114 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
115 errmsg("timeout cannot be negative"));
116
117 timeout = (int64) result;
118 }
119 else if (strcmp(defel->defname, "no_throw") == 0)
120 {
121 if (no_throw_specified)
122 errorConflictingDefElem(defel, pstate);
123
124 no_throw_specified = true;
125
126 throw = !defGetBoolean(defel);
127 }
128 else
129 {
130 ereport(ERROR,
131 errcode(ERRCODE_SYNTAX_ERROR),
132 errmsg("option \"%s\" not recognized",
133 defel->defname),
134 parser_errposition(pstate, defel->location));
135 }
136 }
137
138 /*
139 * We are going to wait for the LSN. We should first care that we don't
140 * hold a snapshot and correspondingly our MyProc->xmin is invalid.
141 * Otherwise, our snapshot could prevent the replay of WAL records
142 * implying a kind of self-deadlock. This is the reason why WAIT FOR is a
143 * command, not a procedure or function.
144 *
145 * At first, we should check there is no active snapshot. According to
146 * PlannedStmtRequiresSnapshot(), even in an atomic context, CallStmt is
147 * processed with a snapshot. Thankfully, we can pop this snapshot,
148 * because PortalRunUtility() can tolerate this.
149 */
150 if (ActiveSnapshotSet())
151 PopActiveSnapshot();
152
153 /*
154 * At second, invalidate a catalog snapshot if any. And we should be done
155 * with the preparation.
156 */
157 InvalidateCatalogSnapshot();
158
159 /* Give up if there is still an active or registered snapshot. */
160 if (HaveRegisteredOrActiveSnapshot())
161 ereport(ERROR,
162 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
163 errmsg("WAIT FOR must be called without an active or registered snapshot"),
164 errdetail("WAIT FOR cannot be executed from a function or procedure, nor within a transaction with an isolation level higher than READ COMMITTED."));
165
166 /*
167 * As the result we should hold no snapshot, and correspondingly our xmin
168 * should be unset.
169 */
170 Assert(MyProc->xmin == InvalidTransactionId);
171
172 /*
173 * Validate that the requested mode matches the current server state.
174 * Primary modes can only be used on a primary.
175 */
176 if (lsnType == WAIT_LSN_TYPE_PRIMARY_FLUSH)
177 {
178 if (RecoveryInProgress())
179 ereport(ERROR,
180 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
181 errmsg("recovery is in progress"),
182 errhint("Waiting for primary_flush can only be done on a primary server. "
183 "Use standby_flush mode on a standby server.")));
184 }
185
186 /* Now wait for the LSN */
187 waitLSNResult = WaitForLSN(lsnType, lsn, timeout);
188
189 /*
190 * Process the result of WaitForLSN(). Throw appropriate error if needed.
191 */
192 switch (waitLSNResult)
193 {
194 case WAIT_LSN_RESULT_SUCCESS:
195 /* Nothing to do on success */
196 result = "success";
197 break;
198
199 case WAIT_LSN_RESULT_TIMEOUT:
200 if (throw)
201 {
202 XLogRecPtr currentLSN = GetCurrentLSNForWaitType(lsnType);
203
204 switch (lsnType)
205 {
206 case WAIT_LSN_TYPE_STANDBY_REPLAY:
207 ereport(ERROR,
208 errcode(ERRCODE_QUERY_CANCELED),
209 errmsg("timed out while waiting for target LSN %X/%08X to be replayed; current standby_replay LSN %X/%08X",
210 LSN_FORMAT_ARGS(lsn),
211 LSN_FORMAT_ARGS(currentLSN)));
212 break;
213
214 case WAIT_LSN_TYPE_STANDBY_WRITE:
215 ereport(ERROR,
216 errcode(ERRCODE_QUERY_CANCELED),
217 errmsg("timed out while waiting for target LSN %X/%08X to be written; current standby_write LSN %X/%08X",
218 LSN_FORMAT_ARGS(lsn),
219 LSN_FORMAT_ARGS(currentLSN)));
220 break;
221
222 case WAIT_LSN_TYPE_STANDBY_FLUSH:
223 ereport(ERROR,
224 errcode(ERRCODE_QUERY_CANCELED),
225 errmsg("timed out while waiting for target LSN %X/%08X to be flushed; current standby_flush LSN %X/%08X",
226 LSN_FORMAT_ARGS(lsn),
227 LSN_FORMAT_ARGS(currentLSN)));
228 break;
229
230 case WAIT_LSN_TYPE_PRIMARY_FLUSH:
231 ereport(ERROR,
232 errcode(ERRCODE_QUERY_CANCELED),
233 errmsg("timed out while waiting for target LSN %X/%08X to be flushed; current primary_flush LSN %X/%08X",
234 LSN_FORMAT_ARGS(lsn),
235 LSN_FORMAT_ARGS(currentLSN)));
236 break;
237
238 default:
239 elog(ERROR, "unexpected wait LSN type %d", lsnType);
240 }
241 }
242 else
243 result = "timeout";
244 break;
245
246 case WAIT_LSN_RESULT_NOT_IN_RECOVERY:
247 if (throw)
248 {
249 if (PromoteIsTriggered())
250 {
251 XLogRecPtr currentLSN = GetCurrentLSNForWaitType(lsnType);
252
253 switch (lsnType)
254 {
255 case WAIT_LSN_TYPE_STANDBY_REPLAY:
256 ereport(ERROR,
257 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
258 errmsg("recovery is not in progress"),
259 errdetail("Recovery ended before target LSN %X/%08X was replayed; last standby_replay LSN %X/%08X.",
260 LSN_FORMAT_ARGS(lsn),
261 LSN_FORMAT_ARGS(currentLSN)));
262 break;
263
264 case WAIT_LSN_TYPE_STANDBY_WRITE:
265 ereport(ERROR,
266 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
267 errmsg("recovery is not in progress"),
268 errdetail("Recovery ended before target LSN %X/%08X was written; last standby_write LSN %X/%08X.",
269 LSN_FORMAT_ARGS(lsn),
270 LSN_FORMAT_ARGS(currentLSN)));
271 break;
272
273 case WAIT_LSN_TYPE_STANDBY_FLUSH:
274 ereport(ERROR,
275 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
276 errmsg("recovery is not in progress"),
277 errdetail("Recovery ended before target LSN %X/%08X was flushed; last standby_flush LSN %X/%08X.",
278 LSN_FORMAT_ARGS(lsn),
279 LSN_FORMAT_ARGS(currentLSN)));
280 break;
281
282 default:
283 elog(ERROR, "unexpected wait LSN type %d", lsnType);
284 }
285 }
286 else
287 {
288 switch (lsnType)
289 {
290 case WAIT_LSN_TYPE_STANDBY_REPLAY:
291 ereport(ERROR,
292 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
293 errmsg("recovery is not in progress"),
294 errhint("Waiting for the standby_replay LSN can only be executed during recovery."));
295 break;
296
297 case WAIT_LSN_TYPE_STANDBY_WRITE:
298 ereport(ERROR,
299 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
300 errmsg("recovery is not in progress"),
301 errhint("Waiting for the standby_write LSN can only be executed during recovery."));
302 break;
303
304 case WAIT_LSN_TYPE_STANDBY_FLUSH:
305 ereport(ERROR,
306 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
307 errmsg("recovery is not in progress"),
308 errhint("Waiting for the standby_flush LSN can only be executed during recovery."));
309 break;
310
311 default:
312 elog(ERROR, "unexpected wait LSN type %d", lsnType);
313 }
314 }
315 }
316 else
317 result = "not in recovery";
318 break;
319 }
320
321 /* need a tuple descriptor representing a single TEXT column */
322 tupdesc = WaitStmtResultDesc(stmt);
323
324 /* prepare for projection of tuples */
325 tstate = begin_tup_output_tupdesc(dest, tupdesc, &TTSOpsVirtual);
326
327 /* Send it */
328 do_text_output_oneline(tstate, result);
329
330 end_tup_output(tstate);
331}
Assert
#define Assert(condition)
Definition c.h:945
int64
int64_t int64
Definition c.h:615
FLOAT8_FITS_IN_INT64
#define FLOAT8_FITS_IN_INT64(num)
Definition c.h:1181
unlikely
#define unlikely(x)
Definition c.h:432
defGetString
char * defGetString(DefElem *def)
Definition define.c:34
defGetBoolean
bool defGetBoolean(DefElem *def)
Definition define.c:93
errorConflictingDefElem
void errorConflictingDefElem(DefElem *defel, ParseState *pstate)
Definition define.c:370
errcode
int errcode(int sqlerrcode)
Definition elog.c:874
_
#define _(x)
Definition elog.c:95
errhint
int errhint(const char *fmt,...) pg_attribute_printf(1
errdetail
int errdetail(const char *fmt,...) pg_attribute_printf(1
ERROR
#define ERROR
Definition elog.h:39
elog
#define elog(elevel,...)
Definition elog.h:226
ereport
#define ereport(elevel,...)
Definition elog.h:150
TTSOpsVirtual
const TupleTableSlotOps TTSOpsVirtual
Definition execTuples.c:84
end_tup_output
void end_tup_output(TupOutputState *tstate)
Definition execTuples.c:2603
begin_tup_output_tupdesc
TupOutputState * begin_tup_output_tupdesc(DestReceiver *dest, TupleDesc tupdesc, const TupleTableSlotOps *tts_ops)
Definition execTuples.c:2525
do_text_output_oneline
#define do_text_output_oneline(tstate, str_to_emit)
Definition executor.h:632
DirectFunctionCall1
#define DirectFunctionCall1(func, arg1)
Definition fmgr.h:684
parse_real
bool parse_real(const char *value, double *result, int flags, const char **hintmsg)
Definition guc.c:2865
GUC_UNIT_MS
#define GUC_UNIT_MS
Definition guc.h:239
stmt
#define stmt
Definition indent_codes.h:59
errmsg
static char * errmsg
Definition oauth_hook_client.c:61
parser_errposition
int parser_errposition(ParseState *pstate, int location)
Definition parse_node.c:106
foreach_node
#define foreach_node(type, var, lst)
Definition pg_list.h:496
pg_lsn_in
Datum pg_lsn_in(PG_FUNCTION_ARGS)
Definition pg_lsn.c:64
DatumGetLSN
static XLogRecPtr DatumGetLSN(Datum X)
Definition pg_lsn.h:25
pg_strcasecmp
int pg_strcasecmp(const char *s1, const char *s2)
Definition pgstrcasecmp.c:32
CStringGetDatum
static Datum CStringGetDatum(const char *X)
Definition postgres.h:370
fb
static int fb(int x)
Definition preproc-init.c:92
ActiveSnapshotSet
bool ActiveSnapshotSet(void)
Definition snapmgr.c:812
HaveRegisteredOrActiveSnapshot
bool HaveRegisteredOrActiveSnapshot(void)
Definition snapmgr.c:1644
PopActiveSnapshot
void PopActiveSnapshot(void)
Definition snapmgr.c:775
InvalidateCatalogSnapshot
void InvalidateCatalogSnapshot(void)
Definition snapmgr.c:455
MyProc
PGPROC * MyProc
Definition proc.c:68
PGPROC::xmin
TransactionId xmin
Definition proc.h:239
InvalidTransactionId
#define InvalidTransactionId
Definition transam.h:31
WaitStmtResultDesc
TupleDesc WaitStmtResultDesc(WaitStmt *stmt)
Definition wait.c:334
RecoveryInProgress
bool RecoveryInProgress(void)
Definition xlog.c:6444
LSN_FORMAT_ARGS
#define LSN_FORMAT_ARGS(lsn)
Definition xlogdefs.h:47
XLogRecPtr
uint64 XLogRecPtr
Definition xlogdefs.h:21
PromoteIsTriggered
bool PromoteIsTriggered(void)
Definition xlogrecovery.c:4399
GetCurrentLSNForWaitType
XLogRecPtr GetCurrentLSNForWaitType(WaitLSNType lsnType)
Definition xlogwait.c:89
WaitForLSN
WaitLSNResult WaitForLSN(WaitLSNType lsnType, XLogRecPtr targetLSN, int64 timeout)
Definition xlogwait.c:376
WaitLSNResult
WaitLSNResult
Definition xlogwait.h:26
WAIT_LSN_RESULT_NOT_IN_RECOVERY
@ WAIT_LSN_RESULT_NOT_IN_RECOVERY
Definition xlogwait.h:28
WAIT_LSN_RESULT_TIMEOUT
@ WAIT_LSN_RESULT_TIMEOUT
Definition xlogwait.h:30
WAIT_LSN_RESULT_SUCCESS
@ WAIT_LSN_RESULT_SUCCESS
Definition xlogwait.h:27
WaitLSNType
WaitLSNType
Definition xlogwait.h:37
WAIT_LSN_TYPE_PRIMARY_FLUSH
@ WAIT_LSN_TYPE_PRIMARY_FLUSH
Definition xlogwait.h:44
WAIT_LSN_TYPE_STANDBY_REPLAY
@ WAIT_LSN_TYPE_STANDBY_REPLAY
Definition xlogwait.h:39
WAIT_LSN_TYPE_STANDBY_FLUSH
@ WAIT_LSN_TYPE_STANDBY_FLUSH
Definition xlogwait.h:41
WAIT_LSN_TYPE_STANDBY_WRITE
@ WAIT_LSN_TYPE_STANDBY_WRITE
Definition xlogwait.h:40

References _, ActiveSnapshotSet(), Assert, begin_tup_output_tupdesc(), CStringGetDatum(), DatumGetLSN(), defGetBoolean(), defGetString(), DirectFunctionCall1, do_text_output_oneline, elog, end_tup_output(), ereport, errcode(), errdetail(), errhint(), errmsg, ERROR, errorConflictingDefElem(), fb(), FLOAT8_FITS_IN_INT64, foreach_node, GetCurrentLSNForWaitType(), GUC_UNIT_MS, HaveRegisteredOrActiveSnapshot(), InvalidateCatalogSnapshot(), InvalidTransactionId, LSN_FORMAT_ARGS, MyProc, parse_real(), parser_errposition(), pg_lsn_in(), pg_strcasecmp(), PopActiveSnapshot(), PromoteIsTriggered(), RecoveryInProgress(), stmt, TTSOpsVirtual, unlikely, WAIT_LSN_RESULT_NOT_IN_RECOVERY, WAIT_LSN_RESULT_SUCCESS, WAIT_LSN_RESULT_TIMEOUT, WAIT_LSN_TYPE_PRIMARY_FLUSH, WAIT_LSN_TYPE_STANDBY_FLUSH, WAIT_LSN_TYPE_STANDBY_REPLAY, WAIT_LSN_TYPE_STANDBY_WRITE, WaitForLSN(), WaitStmtResultDesc(), and PGPROC::xmin.

Referenced by standard_ProcessUtility().

◆ WaitStmtResultDesc()

TupleDesc WaitStmtResultDesc ( WaitStmt *  stmt)

Definition at line 334 of file wait.c.

335{
336 TupleDesc tupdesc;
337
338 /* Need a tuple descriptor representing a single TEXT column */
339 tupdesc = CreateTemplateTupleDesc(1);
340 TupleDescInitEntry(tupdesc, (AttrNumber) 1, "status",
341 TEXTOID, -1, 0);
342 TupleDescFinalize(tupdesc);
343 return tupdesc;
344}
AttrNumber
int16 AttrNumber
Definition attnum.h:21
CreateTemplateTupleDesc
TupleDesc CreateTemplateTupleDesc(int natts)
Definition tupdesc.c:165
TupleDescFinalize
void TupleDescFinalize(TupleDesc tupdesc)
Definition tupdesc.c:508
TupleDescInitEntry
void TupleDescInitEntry(TupleDesc desc, AttrNumber attributeNumber, const char *attributeName, Oid oidtypeid, int32 typmod, int attdim)
Definition tupdesc.c:897

References CreateTemplateTupleDesc(), fb(), TupleDescFinalize(), and TupleDescInitEntry().

Referenced by ExecWaitStmt(), and UtilityTupleDescriptor().