Age Owner Branch data TLA 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-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 : :
31 : : static void pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg);
32 : : static void mq_comm_reset(void);
33 : : static int mq_flush(void);
34 : : static int mq_flush_if_writable(void);
35 : : static bool mq_is_send_pending(void);
36 : : static int mq_putmessage(char msgtype, const char *s, size_t len);
37 : : static 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 : : */
52 : : void
3613 rhaas@postgresql.org 53 :CBC 1388 : pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh)
54 : : {
3963 55 : 1388 : PqCommMethods = &PqCommMqMethods;
56 : 1388 : pq_mq_handle = mqh;
57 : 1388 : whereToSendOutput = DestRemote;
58 : 1388 : FrontendProtocol = PG_PROTOCOL_LATEST;
3613 59 : 1388 : on_dsm_detach(seg, pq_cleanup_redirect_to_shm_mq, (Datum) 0);
60 : 1388 : }
61 : :
62 : : /*
63 : : * When the DSM that contains our shm_mq goes away, we need to stop sending
64 : : * messages to it.
65 : : */
66 : : static void
67 : 1388 : pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg)
68 : : {
35 tgl@sss.pgh.pa.us 69 [ + - ]:GNC 1388 : if (pq_mq_handle != NULL)
70 : : {
71 : 1388 : pfree(pq_mq_handle);
72 : 1388 : pq_mq_handle = NULL;
73 : : }
3613 rhaas@postgresql.org 74 :CBC 1388 : whereToSendOutput = DestNone;
3963 75 : 1388 : }
76 : :
77 : : /*
78 : : * Arrange to SendProcSignal() to the parallel leader each time we transmit
79 : : * message data via the shm_mq.
80 : : */
81 : : void
552 heikki.linnakangas@i 82 : 1388 : pq_set_parallel_leader(pid_t pid, ProcNumber procNumber)
83 : : {
3782 rhaas@postgresql.org 84 [ - + ]: 1388 : Assert(PqCommMethods == &PqCommMqMethods);
1910 andres@anarazel.de 85 : 1388 : pq_mq_parallel_leader_pid = pid;
552 heikki.linnakangas@i 86 : 1388 : pq_mq_parallel_leader_proc_number = procNumber;
3782 rhaas@postgresql.org 87 : 1388 : }
88 : :
89 : : static void
3963 rhaas@postgresql.org 90 :UBC 0 : mq_comm_reset(void)
91 : : {
92 : : /* Nothing to do. */
93 : 0 : }
94 : :
95 : : static int
3963 rhaas@postgresql.org 96 :CBC 11 : mq_flush(void)
97 : : {
98 : : /* Nothing to do. */
99 : 11 : return 0;
100 : : }
101 : :
102 : : static int
3963 rhaas@postgresql.org 103 :UBC 0 : mq_flush_if_writable(void)
104 : : {
105 : : /* Nothing to do. */
106 : 0 : return 0;
107 : : }
108 : :
109 : : static bool
110 : 0 : mq_is_send_pending(void)
111 : : {
112 : : /* There's never anything pending. */
113 : 0 : 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 : : */
121 : : static int
3963 rhaas@postgresql.org 122 :CBC 1383 : 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 [ - + ]: 1383 : if (pq_mq_busy)
136 : : {
2928 tgl@sss.pgh.pa.us 137 [ # # ]:UBC 0 : if (pq_mq_handle != NULL)
138 : : {
139 : 0 : shm_mq_detach(pq_mq_handle);
35 tgl@sss.pgh.pa.us 140 :UNC 0 : pfree(pq_mq_handle);
141 : 0 : pq_mq_handle = NULL;
142 : : }
3963 rhaas@postgresql.org 143 :UBC 0 : 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 : : */
2928 tgl@sss.pgh.pa.us 152 [ - + ]:CBC 1383 : if (pq_mq_handle == NULL)
3613 rhaas@postgresql.org 153 :UBC 0 : return 0;
154 : :
3963 rhaas@postgresql.org 155 :CBC 1383 : pq_mq_busy = true;
156 : :
157 : 1383 : iov[0].data = &msgtype;
158 : 1383 : iov[0].len = 1;
159 : 1383 : iov[1].data = s;
160 : 1383 : 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 : : */
35 tgl@sss.pgh.pa.us 169 [ - + ]:GNC 1387 : Assert(pq_mq_handle != NULL);
1423 rhaas@postgresql.org 170 :CBC 1387 : result = shm_mq_sendv(pq_mq_handle, iov, 2, true, true);
171 : :
1910 andres@anarazel.de 172 [ + - ]: 1387 : if (pq_mq_parallel_leader_pid != 0)
173 : : {
971 akapila@postgresql.o 174 [ + + ]: 1387 : if (IsLogicalParallelApplyWorker())
175 : 5 : SendProcSignal(pq_mq_parallel_leader_pid,
176 : : PROCSIG_PARALLEL_APPLY_MESSAGE,
177 : : pq_mq_parallel_leader_proc_number);
178 : : else
179 : : {
180 [ - + ]: 1382 : Assert(IsParallelWorker());
181 : 1382 : SendProcSignal(pq_mq_parallel_leader_pid,
182 : : PROCSIG_PARALLEL_MESSAGE,
183 : : pq_mq_parallel_leader_proc_number);
184 : : }
185 : : }
186 : :
3782 rhaas@postgresql.org 187 [ + + ]: 1387 : if (result != SHM_MQ_WOULD_BLOCK)
188 : 1383 : break;
189 : :
2479 tmunro@postgresql.or 190 : 4 : (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
191 : : WAIT_EVENT_MESSAGE_QUEUE_PUT_MESSAGE);
3014 andres@anarazel.de 192 : 4 : ResetLatch(MyLatch);
3323 tgl@sss.pgh.pa.us 193 [ - + ]: 4 : CHECK_FOR_INTERRUPTS();
194 : : }
195 : :
3963 rhaas@postgresql.org 196 : 1383 : pq_mq_busy = false;
197 : :
198 [ + + - + ]: 1383 : Assert(result == SHM_MQ_SUCCESS || result == SHM_MQ_DETACHED);
199 [ + + ]: 1383 : if (result != SHM_MQ_SUCCESS)
200 : 3 : return EOF;
201 : 1380 : return 0;
202 : : }
203 : :
204 : : static void
3963 rhaas@postgresql.org 205 :UBC 0 : 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 [ # # ]: 0 : 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 : : */
221 : : void
3963 rhaas@postgresql.org 222 :CBC 8 : pq_parse_errornotice(StringInfo msg, ErrorData *edata)
223 : : {
224 : : /* Initialize edata with reasonable defaults. */
225 [ + - + - : 192 : MemSet(edata, 0, sizeof(ErrorData));
+ - + - +
+ ]
226 : 8 : edata->elevel = ERROR;
227 : 8 : edata->assoc_context = CurrentMemoryContext;
228 : :
229 : : /* Loop over fields and extract each one. */
230 : : for (;;)
231 : 65 : {
3759 bruce@momjian.us 232 : 73 : char code = pq_getmsgbyte(msg);
233 : : const char *value;
234 : :
3963 rhaas@postgresql.org 235 [ + + ]: 73 : if (code == '\0')
236 : : {
237 : 8 : pq_getmsgend(msg);
238 : 8 : break;
239 : : }
3355 240 : 65 : value = pq_getmsgrawstring(msg);
241 : :
3963 242 [ + + + + : 65 : switch (code)
+ + - - -
+ - - - -
- + + +
- ]
243 : : {
244 : 8 : case PG_DIAG_SEVERITY:
245 : : /* ignore, trusting we'll get a nonlocalized version */
3298 tgl@sss.pgh.pa.us 246 : 8 : break;
247 : 8 : case PG_DIAG_SEVERITY_NONLOCALIZED:
3963 rhaas@postgresql.org 248 [ - + ]: 8 : 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 : : */
3298 tgl@sss.pgh.pa.us 255 :UBC 0 : edata->elevel = DEBUG1;
256 : : }
3963 rhaas@postgresql.org 257 [ - + ]:CBC 8 : 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 : : */
3298 tgl@sss.pgh.pa.us 263 :UBC 0 : edata->elevel = LOG;
264 : : }
3963 rhaas@postgresql.org 265 [ - + ]:CBC 8 : else if (strcmp(value, "INFO") == 0)
3963 rhaas@postgresql.org 266 :UBC 0 : edata->elevel = INFO;
3963 rhaas@postgresql.org 267 [ - + ]:CBC 8 : else if (strcmp(value, "NOTICE") == 0)
3963 rhaas@postgresql.org 268 :UBC 0 : edata->elevel = NOTICE;
3963 rhaas@postgresql.org 269 [ - + ]:CBC 8 : else if (strcmp(value, "WARNING") == 0)
3963 rhaas@postgresql.org 270 :UBC 0 : edata->elevel = WARNING;
3963 rhaas@postgresql.org 271 [ + - ]:CBC 8 : else if (strcmp(value, "ERROR") == 0)
272 : 8 : edata->elevel = ERROR;
3963 rhaas@postgresql.org 273 [ # # ]:UBC 0 : else if (strcmp(value, "FATAL") == 0)
274 : 0 : edata->elevel = FATAL;
275 [ # # ]: 0 : else if (strcmp(value, "PANIC") == 0)
276 : 0 : edata->elevel = PANIC;
277 : : else
3298 tgl@sss.pgh.pa.us 278 [ # # ]: 0 : elog(ERROR, "unrecognized error severity: \"%s\"", value);
3963 rhaas@postgresql.org 279 :CBC 8 : break;
280 : 8 : case PG_DIAG_SQLSTATE:
281 [ - + ]: 8 : if (strlen(value) != 5)
3298 tgl@sss.pgh.pa.us 282 [ # # ]:UBC 0 : elog(ERROR, "invalid SQLSTATE: \"%s\"", value);
3963 rhaas@postgresql.org 283 :CBC 8 : edata->sqlerrcode = MAKE_SQLSTATE(value[0], value[1], value[2],
284 : : value[3], value[4]);
285 : 8 : break;
286 : 8 : case PG_DIAG_MESSAGE_PRIMARY:
287 : 8 : edata->message = pstrdup(value);
288 : 8 : break;
289 : 2 : case PG_DIAG_MESSAGE_DETAIL:
290 : 2 : edata->detail = pstrdup(value);
291 : 2 : break;
292 : 2 : case PG_DIAG_MESSAGE_HINT:
293 : 2 : edata->hint = pstrdup(value);
294 : 2 : break;
3963 rhaas@postgresql.org 295 :UBC 0 : case PG_DIAG_STATEMENT_POSITION:
2603 andres@anarazel.de 296 : 0 : edata->cursorpos = pg_strtoint32(value);
3963 rhaas@postgresql.org 297 : 0 : break;
298 : 0 : case PG_DIAG_INTERNAL_POSITION:
2603 andres@anarazel.de 299 : 0 : edata->internalpos = pg_strtoint32(value);
3963 rhaas@postgresql.org 300 : 0 : break;
301 : 0 : case PG_DIAG_INTERNAL_QUERY:
302 : 0 : edata->internalquery = pstrdup(value);
303 : 0 : break;
3963 rhaas@postgresql.org 304 :CBC 5 : case PG_DIAG_CONTEXT:
305 : 5 : edata->context = pstrdup(value);
306 : 5 : break;
3963 rhaas@postgresql.org 307 :UBC 0 : case PG_DIAG_SCHEMA_NAME:
308 : 0 : edata->schema_name = pstrdup(value);
309 : 0 : break;
310 : 0 : case PG_DIAG_TABLE_NAME:
311 : 0 : edata->table_name = pstrdup(value);
312 : 0 : break;
313 : 0 : case PG_DIAG_COLUMN_NAME:
314 : 0 : edata->column_name = pstrdup(value);
315 : 0 : break;
316 : 0 : case PG_DIAG_DATATYPE_NAME:
317 : 0 : edata->datatype_name = pstrdup(value);
318 : 0 : break;
319 : 0 : case PG_DIAG_CONSTRAINT_NAME:
320 : 0 : edata->constraint_name = pstrdup(value);
321 : 0 : break;
3963 rhaas@postgresql.org 322 :CBC 8 : case PG_DIAG_SOURCE_FILE:
323 : 8 : edata->filename = pstrdup(value);
324 : 8 : break;
325 : 8 : case PG_DIAG_SOURCE_LINE:
2603 andres@anarazel.de 326 : 8 : edata->lineno = pg_strtoint32(value);
3963 rhaas@postgresql.org 327 : 8 : break;
328 : 8 : case PG_DIAG_SOURCE_FUNCTION:
329 : 8 : edata->funcname = pstrdup(value);
330 : 8 : break;
3963 rhaas@postgresql.org 331 :UBC 0 : default:
3298 tgl@sss.pgh.pa.us 332 [ # # ]: 0 : elog(ERROR, "unrecognized error field code: %d", (int) code);
333 : : break;
334 : : }
335 : : }
3963 rhaas@postgresql.org 336 :CBC 8 : }
|
