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-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
3910 rhaas@postgresql.org 56 :CBC 2031 : pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh)
57 : : {
4260 58 : 2031 : PqCommMethods = &PqCommMqMethods;
59 : 2031 : pq_mq_handle = mqh;
60 : 2031 : whereToSendOutput = DestRemote;
61 : 2031 : FrontendProtocol = PG_PROTOCOL_LATEST;
3910 62 : 2031 : on_dsm_detach(seg, pq_cleanup_redirect_to_shm_mq, (Datum) 0);
63 : 2031 : }
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 : 2031 : pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg)
71 : : {
332 tgl@sss.pgh.pa.us 72 [ + - ]:GNC 2031 : if (pq_mq_handle != NULL)
73 : : {
74 : 2031 : pfree(pq_mq_handle);
75 : 2031 : pq_mq_handle = NULL;
76 : : }
3910 rhaas@postgresql.org 77 :CBC 2031 : whereToSendOutput = DestNone;
4260 78 : 2031 : }
79 : :
80 : : /*
81 : : * Arrange to SendProcSignal() to the parallel leader each time we transmit
82 : : * message data via the shm_mq.
83 : : */
84 : : void
849 heikki.linnakangas@i 85 : 2031 : pq_set_parallel_leader(pid_t pid, ProcNumber procNumber)
86 : : {
4079 rhaas@postgresql.org 87 [ - + ]: 2031 : Assert(PqCommMethods == &PqCommMqMethods);
2207 andres@anarazel.de 88 : 2031 : pq_mq_parallel_leader_pid = pid;
849 heikki.linnakangas@i 89 : 2031 : pq_mq_parallel_leader_proc_number = procNumber;
4079 rhaas@postgresql.org 90 : 2031 : }
91 : :
92 : : static void
4260 rhaas@postgresql.org 93 :UBC 0 : mq_comm_reset(void)
94 : : {
95 : : /* Nothing to do. */
96 : 0 : }
97 : :
98 : : static int
4260 rhaas@postgresql.org 99 :CBC 16 : mq_flush(void)
100 : : {
101 : : /* Nothing to do. */
102 : 16 : return 0;
103 : : }
104 : :
105 : : static int
4260 rhaas@postgresql.org 106 :UBC 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
4260 rhaas@postgresql.org 125 :CBC 2024 : 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 [ - + ]: 2024 : if (pq_mq_busy)
139 : : {
3225 tgl@sss.pgh.pa.us 140 [ # # ]:UBC 0 : if (pq_mq_handle != NULL)
141 : : {
142 : 0 : shm_mq_detach(pq_mq_handle);
332 tgl@sss.pgh.pa.us 143 :UNC 0 : pq_mq_handle = NULL;
144 : : }
4260 rhaas@postgresql.org 145 :UBC 0 : return EOF;
146 : : }
147 : :
148 : : /*
149 : : * If the message queue is already gone, just ignore the message. This
150 : : * doesn't necessarily indicate a problem; for example, DEBUG messages can
151 : : * be generated late in the shutdown sequence, after all DSMs have already
152 : : * been detached.
153 : : */
3225 tgl@sss.pgh.pa.us 154 [ - + ]:CBC 2024 : if (pq_mq_handle == NULL)
3910 rhaas@postgresql.org 155 :UBC 0 : return 0;
156 : :
4260 rhaas@postgresql.org 157 :CBC 2024 : pq_mq_busy = true;
158 : :
159 : 2024 : iov[0].data = &msgtype;
160 : 2024 : iov[0].len = 1;
161 : 2024 : iov[1].data = s;
162 : 2024 : iov[1].len = len;
163 : :
164 : : for (;;)
165 : : {
166 : : /*
167 : : * Immediately notify the receiver by passing force_flush as true so
168 : : * that the shared memory value is updated before we send the parallel
169 : : * message signal right after this.
170 : : */
332 tgl@sss.pgh.pa.us 171 [ - + ]:GNC 2028 : Assert(pq_mq_handle != NULL);
1720 rhaas@postgresql.org 172 :CBC 2028 : result = shm_mq_sendv(pq_mq_handle, iov, 2, true, true);
173 : :
2207 andres@anarazel.de 174 [ + - ]: 2028 : if (pq_mq_parallel_leader_pid != 0)
175 : : {
1268 akapila@postgresql.o 176 [ + + ]: 2028 : if (IsLogicalParallelApplyWorker())
177 : 7 : SendProcSignal(pq_mq_parallel_leader_pid,
178 : : PROCSIG_PARALLEL_APPLY_MESSAGE,
179 : : pq_mq_parallel_leader_proc_number);
85 alvherre@kurilemu.de 180 [ - + ]:GNC 2021 : else if (AmRepackWorker())
85 alvherre@kurilemu.de 181 :UNC 0 : SendProcSignal(pq_mq_parallel_leader_pid,
182 : : PROCSIG_REPACK_MESSAGE,
183 : : pq_mq_parallel_leader_proc_number);
184 : : else
185 : : {
1268 akapila@postgresql.o 186 [ - + ]:CBC 2021 : Assert(IsParallelWorker());
187 : 2021 : SendProcSignal(pq_mq_parallel_leader_pid,
188 : : PROCSIG_PARALLEL_MESSAGE,
189 : : pq_mq_parallel_leader_proc_number);
190 : : }
191 : : }
192 : :
4079 rhaas@postgresql.org 193 [ + + ]: 2028 : if (result != SHM_MQ_WOULD_BLOCK)
194 : 2024 : break;
195 : :
2776 tmunro@postgresql.or 196 : 4 : (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
197 : : WAIT_EVENT_MESSAGE_QUEUE_PUT_MESSAGE);
3311 andres@anarazel.de 198 : 4 : ResetLatch(MyLatch);
3620 tgl@sss.pgh.pa.us 199 [ - + ]: 4 : CHECK_FOR_INTERRUPTS();
200 : : }
201 : :
4260 rhaas@postgresql.org 202 : 2024 : pq_mq_busy = false;
203 : :
204 [ + + - + ]: 2024 : Assert(result == SHM_MQ_SUCCESS || result == SHM_MQ_DETACHED);
205 [ + + ]: 2024 : if (result != SHM_MQ_SUCCESS)
206 : 5 : return EOF;
207 : 2019 : return 0;
208 : : }
209 : :
210 : : static void
4260 rhaas@postgresql.org 211 :UBC 0 : mq_putmessage_noblock(char msgtype, const char *s, size_t len)
212 : : {
213 : : /*
214 : : * While the shm_mq machinery does support sending a message in
215 : : * non-blocking mode, there's currently no way to try sending beginning to
216 : : * send the message that doesn't also commit us to completing the
217 : : * transmission. This could be improved in the future, but for now we
218 : : * don't need it.
219 : : */
220 [ # # ]: 0 : elog(ERROR, "not currently supported");
221 : : }
222 : :
223 : : /*
224 : : * Parse an ErrorResponse or NoticeResponse payload and populate an ErrorData
225 : : * structure with the results.
226 : : */
227 : : void
4260 rhaas@postgresql.org 228 :CBC 11 : pq_parse_errornotice(StringInfo msg, ErrorData *edata)
229 : : {
230 : : /* Initialize edata with reasonable defaults. */
231 [ + - + - : 264 : MemSet(edata, 0, sizeof(ErrorData));
+ - + - +
+ ]
232 : 11 : edata->elevel = ERROR;
233 : 11 : edata->assoc_context = CurrentMemoryContext;
234 : :
235 : : /* Loop over fields and extract each one. */
236 : : for (;;)
237 : 89 : {
4056 bruce@momjian.us 238 : 100 : char code = pq_getmsgbyte(msg);
239 : : const char *value;
240 : :
4260 rhaas@postgresql.org 241 [ + + ]: 100 : if (code == '\0')
242 : : {
243 : 11 : pq_getmsgend(msg);
244 : 11 : break;
245 : : }
3652 246 : 89 : value = pq_getmsgrawstring(msg);
247 : :
4260 248 [ + + + + : 89 : switch (code)
+ + - - -
+ - - - -
- + + +
- ]
249 : : {
250 : 11 : case PG_DIAG_SEVERITY:
251 : : /* ignore, trusting we'll get a nonlocalized version */
3595 tgl@sss.pgh.pa.us 252 : 11 : break;
253 : 11 : case PG_DIAG_SEVERITY_NONLOCALIZED:
4260 rhaas@postgresql.org 254 [ - + ]: 11 : if (strcmp(value, "DEBUG") == 0)
255 : : {
256 : : /*
257 : : * We can't reconstruct the exact DEBUG level, but
258 : : * presumably it was >= client_min_messages, so select
259 : : * DEBUG1 to ensure we'll pass it on to the client.
260 : : */
3595 tgl@sss.pgh.pa.us 261 :UBC 0 : edata->elevel = DEBUG1;
262 : : }
4260 rhaas@postgresql.org 263 [ - + ]:CBC 11 : else if (strcmp(value, "LOG") == 0)
264 : : {
265 : : /*
266 : : * It can't be LOG_SERVER_ONLY, or the worker wouldn't
267 : : * have sent it to us; so LOG is the correct value.
268 : : */
3595 tgl@sss.pgh.pa.us 269 :UBC 0 : edata->elevel = LOG;
270 : : }
4260 rhaas@postgresql.org 271 [ - + ]:CBC 11 : else if (strcmp(value, "INFO") == 0)
4260 rhaas@postgresql.org 272 :UBC 0 : edata->elevel = INFO;
4260 rhaas@postgresql.org 273 [ - + ]:CBC 11 : else if (strcmp(value, "NOTICE") == 0)
4260 rhaas@postgresql.org 274 :UBC 0 : edata->elevel = NOTICE;
4260 rhaas@postgresql.org 275 [ - + ]:CBC 11 : else if (strcmp(value, "WARNING") == 0)
4260 rhaas@postgresql.org 276 :UBC 0 : edata->elevel = WARNING;
4260 rhaas@postgresql.org 277 [ + - ]:CBC 11 : else if (strcmp(value, "ERROR") == 0)
278 : 11 : edata->elevel = ERROR;
4260 rhaas@postgresql.org 279 [ # # ]:UBC 0 : else if (strcmp(value, "FATAL") == 0)
280 : 0 : edata->elevel = FATAL;
281 [ # # ]: 0 : else if (strcmp(value, "PANIC") == 0)
282 : 0 : edata->elevel = PANIC;
283 : : else
3595 tgl@sss.pgh.pa.us 284 [ # # ]: 0 : elog(ERROR, "unrecognized error severity: \"%s\"", value);
4260 rhaas@postgresql.org 285 :CBC 11 : break;
286 : 11 : case PG_DIAG_SQLSTATE:
287 [ - + ]: 11 : if (strlen(value) != 5)
3595 tgl@sss.pgh.pa.us 288 [ # # ]:UBC 0 : elog(ERROR, "invalid SQLSTATE: \"%s\"", value);
4260 rhaas@postgresql.org 289 :CBC 11 : edata->sqlerrcode = MAKE_SQLSTATE(value[0], value[1], value[2],
290 : : value[3], value[4]);
291 : 11 : break;
292 : 11 : case PG_DIAG_MESSAGE_PRIMARY:
293 : 11 : edata->message = pstrdup(value);
294 : 11 : break;
295 : 2 : case PG_DIAG_MESSAGE_DETAIL:
296 : 2 : edata->detail = pstrdup(value);
297 : 2 : break;
298 : 3 : case PG_DIAG_MESSAGE_HINT:
299 : 3 : edata->hint = pstrdup(value);
300 : 3 : break;
4260 rhaas@postgresql.org 301 :UBC 0 : case PG_DIAG_STATEMENT_POSITION:
2900 andres@anarazel.de 302 : 0 : edata->cursorpos = pg_strtoint32(value);
4260 rhaas@postgresql.org 303 : 0 : break;
304 : 0 : case PG_DIAG_INTERNAL_POSITION:
2900 andres@anarazel.de 305 : 0 : edata->internalpos = pg_strtoint32(value);
4260 rhaas@postgresql.org 306 : 0 : break;
307 : 0 : case PG_DIAG_INTERNAL_QUERY:
308 : 0 : edata->internalquery = pstrdup(value);
309 : 0 : break;
4260 rhaas@postgresql.org 310 :CBC 7 : case PG_DIAG_CONTEXT:
311 : 7 : edata->context = pstrdup(value);
312 : 7 : break;
4260 rhaas@postgresql.org 313 :UBC 0 : case PG_DIAG_SCHEMA_NAME:
314 : 0 : edata->schema_name = pstrdup(value);
315 : 0 : break;
316 : 0 : case PG_DIAG_TABLE_NAME:
317 : 0 : edata->table_name = pstrdup(value);
318 : 0 : break;
319 : 0 : case PG_DIAG_COLUMN_NAME:
320 : 0 : edata->column_name = pstrdup(value);
321 : 0 : break;
322 : 0 : case PG_DIAG_DATATYPE_NAME:
323 : 0 : edata->datatype_name = pstrdup(value);
324 : 0 : break;
325 : 0 : case PG_DIAG_CONSTRAINT_NAME:
326 : 0 : edata->constraint_name = pstrdup(value);
327 : 0 : break;
4260 rhaas@postgresql.org 328 :CBC 11 : case PG_DIAG_SOURCE_FILE:
329 : 11 : edata->filename = pstrdup(value);
330 : 11 : break;
331 : 11 : case PG_DIAG_SOURCE_LINE:
2900 andres@anarazel.de 332 : 11 : edata->lineno = pg_strtoint32(value);
4260 rhaas@postgresql.org 333 : 11 : break;
334 : 11 : case PG_DIAG_SOURCE_FUNCTION:
335 : 11 : edata->funcname = pstrdup(value);
336 : 11 : break;
4260 rhaas@postgresql.org 337 :UBC 0 : default:
203 peter@eisentraut.org 338 [ # # ]:UNC 0 : elog(ERROR, "unrecognized error field code: %d", code);
339 : : break;
340 : : }
341 : : }
4260 rhaas@postgresql.org 342 :CBC 11 : }
|
