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