PostgreSQL Source Code: src/backend/libpq/pqmq.c Source File

PostgreSQL Source Code git master
pqmq.c
Go to the documentation of this file.
1/*-------------------------------------------------------------------------
2 *
3 * pqmq.c
4 * Use the frontend/backend protocol for communication over a shm_mq
5 *
6 * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
7 * Portions Copyright (c) 1994, Regents of the University of California
8 *
9 * src/backend/libpq/pqmq.c
10 *
11 *-------------------------------------------------------------------------
12 */
13
14#include "postgres.h"
15
16#include "access/parallel.h"
17#include "libpq/libpq.h"
18#include "libpq/pqformat.h"
19#include "libpq/pqmq.h"
20#include "miscadmin.h"
21#include "pgstat.h"
22#include "replication/logicalworker.h"
23#include "tcop/tcopprot.h"
24#include "utils/builtins.h"
25
26 static shm_mq_handle *pq_mq_handle = NULL;
27 static bool pq_mq_busy = false;
28 static pid_t pq_mq_parallel_leader_pid = 0;
29 static ProcNumber pq_mq_parallel_leader_proc_number = INVALID_PROC_NUMBER;
30
31static void pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg);
32static void mq_comm_reset(void);
33static int mq_flush(void);
34static int mq_flush_if_writable(void);
35static bool mq_is_send_pending(void);
36static int mq_putmessage(char msgtype, const char *s, size_t len);
37static void mq_putmessage_noblock(char msgtype, const char *s, size_t len);
38
39 static const PQcommMethods PqCommMqMethods = {
40 .comm_reset = mq_comm_reset,
41 .flush = mq_flush,
42 .flush_if_writable = mq_flush_if_writable,
43 .is_send_pending = mq_is_send_pending,
44 .putmessage = mq_putmessage,
45 .putmessage_noblock = mq_putmessage_noblock
46};
47
48/*
49 * Arrange to redirect frontend/backend protocol messages to a shared-memory
50 * message queue.
51 */
52void
53 pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh)
54{
55 PqCommMethods = &PqCommMqMethods;
56 pq_mq_handle = mqh;
57 whereToSendOutput = DestRemote;
58 FrontendProtocol = PG_PROTOCOL_LATEST;
59 on_dsm_detach(seg, pq_cleanup_redirect_to_shm_mq, (Datum) 0);
60}
61
62/*
63 * When the DSM that contains our shm_mq goes away, we need to stop sending
64 * messages to it.
65 */
66static void
67 pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg)
68{
69 if (pq_mq_handle != NULL)
70 {
71 pfree(pq_mq_handle);
72 pq_mq_handle = NULL;
73 }
74 whereToSendOutput = DestNone;
75}
76
77/*
78 * Arrange to SendProcSignal() to the parallel leader each time we transmit
79 * message data via the shm_mq.
80 */
81void
82 pq_set_parallel_leader(pid_t pid, ProcNumber procNumber)
83{
84 Assert(PqCommMethods == &PqCommMqMethods);
85 pq_mq_parallel_leader_pid = pid;
86 pq_mq_parallel_leader_proc_number = procNumber;
87}
88
89static void
90 mq_comm_reset(void)
91{
92 /* Nothing to do. */
93}
94
95static int
96 mq_flush(void)
97{
98 /* Nothing to do. */
99 return 0;
100}
101
102static int
103 mq_flush_if_writable(void)
104{
105 /* Nothing to do. */
106 return 0;
107}
108
109static bool
110 mq_is_send_pending(void)
111{
112 /* There's never anything pending. */
113 return 0;
114}
115
116/*
117 * Transmit a libpq protocol message to the shared memory message queue
118 * selected via pq_mq_handle. We don't include a length word, because the
119 * receiver will know the length of the message from shm_mq_receive().
120 */
121static int
122 mq_putmessage(char msgtype, const char *s, size_t len)
123{
124 shm_mq_iovec iov[2];
125 shm_mq_result result;
126
127 /*
128 * If we're sending a message, and we have to wait because the queue is
129 * full, and then we get interrupted, and that interrupt results in trying
130 * to send another message, we respond by detaching the queue. There's no
131 * way to return to the original context, but even if there were, just
132 * queueing the message would amount to indefinitely postponing the
133 * response to the interrupt. So we do this instead.
134 */
135 if (pq_mq_busy)
136 {
137 if (pq_mq_handle != NULL)
138 {
139 shm_mq_detach(pq_mq_handle);
140 pfree(pq_mq_handle);
141 pq_mq_handle = NULL;
142 }
143 return EOF;
144 }
145
146 /*
147 * If the message queue is already gone, just ignore the message. This
148 * doesn't necessarily indicate a problem; for example, DEBUG messages can
149 * be generated late in the shutdown sequence, after all DSMs have already
150 * been detached.
151 */
152 if (pq_mq_handle == NULL)
153 return 0;
154
155 pq_mq_busy = true;
156
157 iov[0].data = &msgtype;
158 iov[0].len = 1;
159 iov[1].data = s;
160 iov[1].len = len;
161
162 for (;;)
163 {
164 /*
165 * Immediately notify the receiver by passing force_flush as true so
166 * that the shared memory value is updated before we send the parallel
167 * message signal right after this.
168 */
169 Assert(pq_mq_handle != NULL);
170 result = shm_mq_sendv(pq_mq_handle, iov, 2, true, true);
171
172 if (pq_mq_parallel_leader_pid != 0)
173 {
174 if (IsLogicalParallelApplyWorker())
175 SendProcSignal(pq_mq_parallel_leader_pid,
176 PROCSIG_PARALLEL_APPLY_MESSAGE,
177 pq_mq_parallel_leader_proc_number);
178 else
179 {
180 Assert(IsParallelWorker());
181 SendProcSignal(pq_mq_parallel_leader_pid,
182 PROCSIG_PARALLEL_MESSAGE,
183 pq_mq_parallel_leader_proc_number);
184 }
185 }
186
187 if (result != SHM_MQ_WOULD_BLOCK)
188 break;
189
190 (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
191 WAIT_EVENT_MESSAGE_QUEUE_PUT_MESSAGE);
192 ResetLatch(MyLatch);
193 CHECK_FOR_INTERRUPTS();
194 }
195
196 pq_mq_busy = false;
197
198 Assert(result == SHM_MQ_SUCCESS || result == SHM_MQ_DETACHED);
199 if (result != SHM_MQ_SUCCESS)
200 return EOF;
201 return 0;
202}
203
204static void
205 mq_putmessage_noblock(char msgtype, const char *s, size_t len)
206{
207 /*
208 * While the shm_mq machinery does support sending a message in
209 * non-blocking mode, there's currently no way to try sending beginning to
210 * send the message that doesn't also commit us to completing the
211 * transmission. This could be improved in the future, but for now we
212 * don't need it.
213 */
214 elog(ERROR, "not currently supported");
215}
216
217/*
218 * Parse an ErrorResponse or NoticeResponse payload and populate an ErrorData
219 * structure with the results.
220 */
221void
222 pq_parse_errornotice(StringInfo msg, ErrorData *edata)
223{
224 /* Initialize edata with reasonable defaults. */
225 MemSet(edata, 0, sizeof(ErrorData));
226 edata->elevel = ERROR;
227 edata->assoc_context = CurrentMemoryContext;
228
229 /* Loop over fields and extract each one. */
230 for (;;)
231 {
232 char code = pq_getmsgbyte(msg);
233 const char *value;
234
235 if (code == '0円')
236 {
237 pq_getmsgend(msg);
238 break;
239 }
240 value = pq_getmsgrawstring(msg);
241
242 switch (code)
243 {
244 case PG_DIAG_SEVERITY:
245 /* ignore, trusting we'll get a nonlocalized version */
246 break;
247 case PG_DIAG_SEVERITY_NONLOCALIZED:
248 if (strcmp(value, "DEBUG") == 0)
249 {
250 /*
251 * We can't reconstruct the exact DEBUG level, but
252 * presumably it was >= client_min_messages, so select
253 * DEBUG1 to ensure we'll pass it on to the client.
254 */
255 edata->elevel = DEBUG1;
256 }
257 else if (strcmp(value, "LOG") == 0)
258 {
259 /*
260 * It can't be LOG_SERVER_ONLY, or the worker wouldn't
261 * have sent it to us; so LOG is the correct value.
262 */
263 edata->elevel = LOG;
264 }
265 else if (strcmp(value, "INFO") == 0)
266 edata->elevel = INFO;
267 else if (strcmp(value, "NOTICE") == 0)
268 edata->elevel = NOTICE;
269 else if (strcmp(value, "WARNING") == 0)
270 edata->elevel = WARNING;
271 else if (strcmp(value, "ERROR") == 0)
272 edata->elevel = ERROR;
273 else if (strcmp(value, "FATAL") == 0)
274 edata->elevel = FATAL;
275 else if (strcmp(value, "PANIC") == 0)
276 edata->elevel = PANIC;
277 else
278 elog(ERROR, "unrecognized error severity: \"%s\"", value);
279 break;
280 case PG_DIAG_SQLSTATE:
281 if (strlen(value) != 5)
282 elog(ERROR, "invalid SQLSTATE: \"%s\"", value);
283 edata->sqlerrcode = MAKE_SQLSTATE(value[0], value[1], value[2],
284 value[3], value[4]);
285 break;
286 case PG_DIAG_MESSAGE_PRIMARY:
287 edata->message = pstrdup(value);
288 break;
289 case PG_DIAG_MESSAGE_DETAIL:
290 edata->detail = pstrdup(value);
291 break;
292 case PG_DIAG_MESSAGE_HINT:
293 edata->hint = pstrdup(value);
294 break;
295 case PG_DIAG_STATEMENT_POSITION:
296 edata->cursorpos = pg_strtoint32(value);
297 break;
298 case PG_DIAG_INTERNAL_POSITION:
299 edata->internalpos = pg_strtoint32(value);
300 break;
301 case PG_DIAG_INTERNAL_QUERY:
302 edata->internalquery = pstrdup(value);
303 break;
304 case PG_DIAG_CONTEXT:
305 edata->context = pstrdup(value);
306 break;
307 case PG_DIAG_SCHEMA_NAME:
308 edata->schema_name = pstrdup(value);
309 break;
310 case PG_DIAG_TABLE_NAME:
311 edata->table_name = pstrdup(value);
312 break;
313 case PG_DIAG_COLUMN_NAME:
314 edata->column_name = pstrdup(value);
315 break;
316 case PG_DIAG_DATATYPE_NAME:
317 edata->datatype_name = pstrdup(value);
318 break;
319 case PG_DIAG_CONSTRAINT_NAME:
320 edata->constraint_name = pstrdup(value);
321 break;
322 case PG_DIAG_SOURCE_FILE:
323 edata->filename = pstrdup(value);
324 break;
325 case PG_DIAG_SOURCE_LINE:
326 edata->lineno = pg_strtoint32(value);
327 break;
328 case PG_DIAG_SOURCE_FUNCTION:
329 edata->funcname = pstrdup(value);
330 break;
331 default:
332 elog(ERROR, "unrecognized error field code: %d", (int) code);
333 break;
334 }
335 }
336}
bool IsLogicalParallelApplyWorker(void)
Definition: worker.c:5970
#define MemSet(start, val, len)
Definition: c.h:1019
@ DestRemote
Definition: dest.h:89
@ DestNone
Definition: dest.h:87
void on_dsm_detach(dsm_segment *seg, on_dsm_detach_callback function, Datum arg)
Definition: dsm.c:1132
#define LOG
Definition: elog.h:31
#define FATAL
Definition: elog.h:41
#define WARNING
Definition: elog.h:36
#define PANIC
Definition: elog.h:42
#define DEBUG1
Definition: elog.h:30
#define ERROR
Definition: elog.h:39
#define MAKE_SQLSTATE(ch1, ch2, ch3, ch4, ch5)
Definition: elog.h:56
#define elog(elevel,...)
Definition: elog.h:226
#define NOTICE
Definition: elog.h:35
#define INFO
Definition: elog.h:34
ProtocolVersion FrontendProtocol
Definition: globals.c:30
struct Latch * MyLatch
Definition: globals.c:63
Assert(PointerIsAligned(start, uint64))
#define IsParallelWorker()
Definition: parallel.h:60
static struct @169 value
void ResetLatch(Latch *latch)
Definition: latch.c:374
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:172
char * pstrdup(const char *in)
Definition: mcxt.c:1759
void pfree(void *pointer)
Definition: mcxt.c:1594
MemoryContext CurrentMemoryContext
Definition: mcxt.c:160
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
int32 pg_strtoint32(const char *s)
Definition: numutils.c:383
void * arg
const void size_t len
CommandDest whereToSendOutput
Definition: postgres.c:91
uint64_t Datum
Definition: postgres.h:70
#define PG_DIAG_INTERNAL_QUERY
Definition: postgres_ext.h:63
#define PG_DIAG_SCHEMA_NAME
Definition: postgres_ext.h:65
#define PG_DIAG_CONSTRAINT_NAME
Definition: postgres_ext.h:69
#define PG_DIAG_DATATYPE_NAME
Definition: postgres_ext.h:68
#define PG_DIAG_SOURCE_LINE
Definition: postgres_ext.h:71
#define PG_DIAG_STATEMENT_POSITION
Definition: postgres_ext.h:61
#define PG_DIAG_SOURCE_FILE
Definition: postgres_ext.h:70
#define PG_DIAG_MESSAGE_HINT
Definition: postgres_ext.h:60
#define PG_DIAG_SQLSTATE
Definition: postgres_ext.h:57
#define PG_DIAG_SEVERITY_NONLOCALIZED
Definition: postgres_ext.h:56
#define PG_DIAG_TABLE_NAME
Definition: postgres_ext.h:66
#define PG_DIAG_MESSAGE_PRIMARY
Definition: postgres_ext.h:58
#define PG_DIAG_COLUMN_NAME
Definition: postgres_ext.h:67
#define PG_DIAG_MESSAGE_DETAIL
Definition: postgres_ext.h:59
#define PG_DIAG_CONTEXT
Definition: postgres_ext.h:64
#define PG_DIAG_SEVERITY
Definition: postgres_ext.h:55
#define PG_DIAG_SOURCE_FUNCTION
Definition: postgres_ext.h:72
#define PG_DIAG_INTERNAL_POSITION
Definition: postgres_ext.h:62
const PQcommMethods * PqCommMethods
Definition: pqcomm.c:164
#define PG_PROTOCOL_LATEST
Definition: pqcomm.h:97
void pq_getmsgend(StringInfo msg)
Definition: pqformat.c:635
int pq_getmsgbyte(StringInfo msg)
Definition: pqformat.c:399
const char * pq_getmsgrawstring(StringInfo msg)
Definition: pqformat.c:608
static int mq_flush(void)
Definition: pqmq.c:96
static bool pq_mq_busy
Definition: pqmq.c:27
static void pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg)
Definition: pqmq.c:67
static const PQcommMethods PqCommMqMethods
Definition: pqmq.c:39
static ProcNumber pq_mq_parallel_leader_proc_number
Definition: pqmq.c:29
void pq_set_parallel_leader(pid_t pid, ProcNumber procNumber)
Definition: pqmq.c:82
static int mq_putmessage(char msgtype, const char *s, size_t len)
Definition: pqmq.c:122
static bool mq_is_send_pending(void)
Definition: pqmq.c:110
static pid_t pq_mq_parallel_leader_pid
Definition: pqmq.c:28
void pq_parse_errornotice(StringInfo msg, ErrorData *edata)
Definition: pqmq.c:222
static shm_mq_handle * pq_mq_handle
Definition: pqmq.c:26
static void mq_putmessage_noblock(char msgtype, const char *s, size_t len)
Definition: pqmq.c:205
void pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh)
Definition: pqmq.c:53
static void mq_comm_reset(void)
Definition: pqmq.c:90
static int mq_flush_if_writable(void)
Definition: pqmq.c:103
#define INVALID_PROC_NUMBER
Definition: procnumber.h:26
int ProcNumber
Definition: procnumber.h:24
int SendProcSignal(pid_t pid, ProcSignalReason reason, ProcNumber procNumber)
Definition: procsignal.c:284
@ PROCSIG_PARALLEL_MESSAGE
Definition: procsignal.h:34
@ PROCSIG_PARALLEL_APPLY_MESSAGE
Definition: procsignal.h:38
void shm_mq_detach(shm_mq_handle *mqh)
Definition: shm_mq.c:843
shm_mq_result shm_mq_sendv(shm_mq_handle *mqh, shm_mq_iovec *iov, int iovcnt, bool nowait, bool force_flush)
Definition: shm_mq.c:361
shm_mq_result
Definition: shm_mq.h:37
@ SHM_MQ_SUCCESS
Definition: shm_mq.h:38
@ SHM_MQ_WOULD_BLOCK
Definition: shm_mq.h:39
@ SHM_MQ_DETACHED
Definition: shm_mq.h:40
Definition: elog.h:420
int internalpos
Definition: elog.h:445
char * schema_name
Definition: elog.h:439
char * context
Definition: elog.h:436
char * internalquery
Definition: elog.h:446
int sqlerrcode
Definition: elog.h:431
struct MemoryContextData * assoc_context
Definition: elog.h:450
const char * filename
Definition: elog.h:426
int elevel
Definition: elog.h:421
char * datatype_name
Definition: elog.h:442
char * detail
Definition: elog.h:433
const char * funcname
Definition: elog.h:428
char * table_name
Definition: elog.h:440
char * message
Definition: elog.h:432
int lineno
Definition: elog.h:427
char * hint
Definition: elog.h:435
char * constraint_name
Definition: elog.h:443
int cursorpos
Definition: elog.h:444
char * column_name
Definition: elog.h:441
void(* comm_reset)(void)
Definition: libpq.h:35
Definition: dsm.c:67
Definition: shm_mq.c:138
Definition: shm_mq.h:30
const char * data
Definition: shm_mq.h:31
Size len
Definition: shm_mq.h:32
#define WL_EXIT_ON_PM_DEATH
Definition: waiteventset.h:39
#define WL_LATCH_SET
Definition: waiteventset.h:34

AltStyle によって変換されたページ (->オリジナル) /