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