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-2021, 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 "libpq/libpq.h"
17 #include "libpq/pqformat.h"
18 #include "libpq/pqmq.h"
19 #include "miscadmin.h"
20 #include "pgstat.h"
21 #include "tcop/tcopprot.h"
22 #include "utils/builtins.h"
23 
25 static bool pq_mq_busy = false;
26 static pid_t pq_mq_parallel_leader_pid = 0;
28 
30 static void mq_comm_reset(void);
31 static int mq_flush(void);
32 static int mq_flush_if_writable(void);
33 static bool mq_is_send_pending(void);
34 static int mq_putmessage(char msgtype, const char *s, size_t len);
35 static void mq_putmessage_noblock(char msgtype, const char *s, size_t len);
36 
39  mq_flush,
44 };
45 
46 /*
47  * Arrange to redirect frontend/backend protocol messages to a shared-memory
48  * message queue.
49  */
50 void
52 {
54  pq_mq_handle = mqh;
58 }
59 
60 /*
61  * When the DSM that contains our shm_mq goes away, we need to stop sending
62  * messages to it.
63  */
64 static void
66 {
67  pq_mq_handle = NULL;
69 }
70 
71 /*
72  * Arrange to SendProcSignal() to the parallel leader each time we transmit
73  * message data via the shm_mq.
74  */
75 void
76 pq_set_parallel_leader(pid_t pid, BackendId backend_id)
77 {
78  Assert(PqCommMethods == &PqCommMqMethods);
81 }
82 
83 static void
85 {
86  /* Nothing to do. */
87 }
88 
89 static int
90 mq_flush(void)
91 {
92  /* Nothing to do. */
93  return 0;
94 }
95 
96 static int
98 {
99  /* Nothing to do. */
100  return 0;
101 }
102 
103 static bool
105 {
106  /* There's never anything pending. */
107  return 0;
108 }
109 
110 /*
111  * Transmit a libpq protocol message to the shared memory message queue
112  * selected via pq_mq_handle. We don't include a length word, because the
113  * receiver will know the length of the message from shm_mq_receive().
114  */
115 static int
116 mq_putmessage(char msgtype, const char *s, size_t len)
117 {
118  shm_mq_iovec iov[2];
119  shm_mq_result result;
120 
121  /*
122  * If we're sending a message, and we have to wait because the queue is
123  * full, and then we get interrupted, and that interrupt results in trying
124  * to send another message, we respond by detaching the queue. There's no
125  * way to return to the original context, but even if there were, just
126  * queueing the message would amount to indefinitely postponing the
127  * response to the interrupt. So we do this instead.
128  */
129  if (pq_mq_busy)
130  {
131  if (pq_mq_handle != NULL)
132  shm_mq_detach(pq_mq_handle);
133  pq_mq_handle = NULL;
134  return EOF;
135  }
136 
137  /*
138  * If the message queue is already gone, just ignore the message. This
139  * doesn't necessarily indicate a problem; for example, DEBUG messages can
140  * be generated late in the shutdown sequence, after all DSMs have already
141  * been detached.
142  */
143  if (pq_mq_handle == NULL)
144  return 0;
145 
146  pq_mq_busy = true;
147 
148  iov[0].data = &msgtype;
149  iov[0].len = 1;
150  iov[1].data = s;
151  iov[1].len = len;
152 
153  Assert(pq_mq_handle != NULL);
154 
155  for (;;)
156  {
157  /*
158  * Immediately notify the receiver by passing force_flush as true so
159  * that the shared memory value is updated before we send the parallel
160  * message signal right after this.
161  */
162  result = shm_mq_sendv(pq_mq_handle, iov, 2, true, true);
163 
164  if (pq_mq_parallel_leader_pid != 0)
168 
169  if (result != SHM_MQ_WOULD_BLOCK)
170  break;
171 
176  }
177 
178  pq_mq_busy = false;
179 
180  Assert(result == SHM_MQ_SUCCESS || result == SHM_MQ_DETACHED);
181  if (result != SHM_MQ_SUCCESS)
182  return EOF;
183  return 0;
184 }
185 
186 static void
187 mq_putmessage_noblock(char msgtype, const char *s, size_t len)
188 {
189  /*
190  * While the shm_mq machinery does support sending a message in
191  * non-blocking mode, there's currently no way to try sending beginning to
192  * send the message that doesn't also commit us to completing the
193  * transmission. This could be improved in the future, but for now we
194  * don't need it.
195  */
196  elog(ERROR, "not currently supported");
197 }
198 
199 /*
200  * Parse an ErrorResponse or NoticeResponse payload and populate an ErrorData
201  * structure with the results.
202  */
203 void
205 {
206  /* Initialize edata with reasonable defaults. */
207  MemSet(edata, 0, sizeof(ErrorData));
208  edata->elevel = ERROR;
210 
211  /* Loop over fields and extract each one. */
212  for (;;)
213  {
214  char code = pq_getmsgbyte(msg);
215  const char *value;
216 
217  if (code == '\0')
218  {
219  pq_getmsgend(msg);
220  break;
221  }
222  value = pq_getmsgrawstring(msg);
223 
224  switch (code)
225  {
226  case PG_DIAG_SEVERITY:
227  /* ignore, trusting we'll get a nonlocalized version */
228  break;
230  if (strcmp(value, "DEBUG") == 0)
231  {
232  /*
233  * We can't reconstruct the exact DEBUG level, but
234  * presumably it was >= client_min_messages, so select
235  * DEBUG1 to ensure we'll pass it on to the client.
236  */
237  edata->elevel = DEBUG1;
238  }
239  else if (strcmp(value, "LOG") == 0)
240  {
241  /*
242  * It can't be LOG_SERVER_ONLY, or the worker wouldn't
243  * have sent it to us; so LOG is the correct value.
244  */
245  edata->elevel = LOG;
246  }
247  else if (strcmp(value, "INFO") == 0)
248  edata->elevel = INFO;
249  else if (strcmp(value, "NOTICE") == 0)
250  edata->elevel = NOTICE;
251  else if (strcmp(value, "WARNING") == 0)
252  edata->elevel = WARNING;
253  else if (strcmp(value, "ERROR") == 0)
254  edata->elevel = ERROR;
255  else if (strcmp(value, "FATAL") == 0)
256  edata->elevel = FATAL;
257  else if (strcmp(value, "PANIC") == 0)
258  edata->elevel = PANIC;
259  else
260  elog(ERROR, "unrecognized error severity: \"%s\"", value);
261  break;
262  case PG_DIAG_SQLSTATE:
263  if (strlen(value) != 5)
264  elog(ERROR, "invalid SQLSTATE: \"%s\"", value);
265  edata->sqlerrcode = MAKE_SQLSTATE(value[0], value[1], value[2],
266  value[3], value[4]);
267  break;
269  edata->message = pstrdup(value);
270  break;
272  edata->detail = pstrdup(value);
273  break;
275  edata->hint = pstrdup(value);
276  break;
278  edata->cursorpos = pg_strtoint32(value);
279  break;
281  edata->internalpos = pg_strtoint32(value);
282  break;
284  edata->internalquery = pstrdup(value);
285  break;
286  case PG_DIAG_CONTEXT:
287  edata->context = pstrdup(value);
288  break;
289  case PG_DIAG_SCHEMA_NAME:
290  edata->schema_name = pstrdup(value);
291  break;
292  case PG_DIAG_TABLE_NAME:
293  edata->table_name = pstrdup(value);
294  break;
295  case PG_DIAG_COLUMN_NAME:
296  edata->column_name = pstrdup(value);
297  break;
299  edata->datatype_name = pstrdup(value);
300  break;
302  edata->constraint_name = pstrdup(value);
303  break;
304  case PG_DIAG_SOURCE_FILE:
305  edata->filename = pstrdup(value);
306  break;
307  case PG_DIAG_SOURCE_LINE:
308  edata->lineno = pg_strtoint32(value);
309  break;
311  edata->funcname = pstrdup(value);
312  break;
313  default:
314  elog(ERROR, "unrecognized error field code: %d", (int) code);
315  break;
316  }
317  }
318 }
char * schema_name
Definition: elog.h:389
#define DEBUG1
Definition: elog.h:25
static shm_mq_handle * pq_mq_handle
Definition: pqmq.c:24
#define PG_DIAG_MESSAGE_PRIMARY
Definition: postgres_ext.h:58
#define PG_DIAG_SCHEMA_NAME
Definition: postgres_ext.h:65
static int mq_putmessage(char msgtype, const char *s, size_t len)
Definition: pqmq.c:116
int sqlerrcode
Definition: elog.h:381
#define PG_DIAG_MESSAGE_DETAIL
Definition: postgres_ext.h:59
void shm_mq_detach(shm_mq_handle *mqh)
Definition: shm_mq.c:842
#define PG_DIAG_COLUMN_NAME
Definition: postgres_ext.h:67
#define MAKE_SQLSTATE(ch1, ch2, ch3, ch4, ch5)
Definition: elog.h:64
const char * funcname
Definition: elog.h:378
static int mq_flush_if_writable(void)
Definition: pqmq.c:97
#define PG_PROTOCOL_LATEST
Definition: pqcomm.h:123
char * pstrdup(const char *in)
Definition: mcxt.c:1299
#define PG_DIAG_TABLE_NAME
Definition: postgres_ext.h:66
static void pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg)
Definition: pqmq.c:65
const PQcommMethods * PqCommMethods
Definition: pqcomm.c:166
int lineno
Definition: elog.h:377
#define PG_DIAG_SOURCE_LINE
Definition: postgres_ext.h:71
#define MemSet(start, val, len)
Definition: c.h:1008
#define INFO
Definition: elog.h:33
char * internalquery
Definition: elog.h:396
#define LOG
Definition: elog.h:26
#define PG_DIAG_INTERNAL_POSITION
Definition: postgres_ext.h:62
const char * pq_getmsgrawstring(StringInfo msg)
Definition: pqformat.c:610
void on_dsm_detach(dsm_segment *seg, on_dsm_detach_callback function, Datum arg)
Definition: dsm.c:1096
#define PG_DIAG_SQLSTATE
Definition: postgres_ext.h:57
static const PQcommMethods PqCommMqMethods
Definition: pqmq.c:37
#define PANIC
Definition: elog.h:50
void ResetLatch(Latch *latch)
Definition: latch.c:660
Size len
Definition: shm_mq.h:32
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:452
static void mq_putmessage_noblock(char msgtype, const char *s, size_t len)
Definition: pqmq.c:187
int cursorpos
Definition: elog.h:394
int SendProcSignal(pid_t pid, ProcSignalReason reason, BackendId backendId)
Definition: procsignal.c:261
#define PG_DIAG_SOURCE_FILE
Definition: postgres_ext.h:70
const char * data
Definition: shm_mq.h:31
#define ERROR
Definition: elog.h:46
const char * filename
Definition: elog.h:376
static void mq_comm_reset(void)
Definition: pqmq.c:84
void pq_parse_errornotice(StringInfo msg, ErrorData *edata)
Definition: pqmq.c:204
#define FATAL
Definition: elog.h:49
Definition: dest.h:89
char * table_name
Definition: elog.h:390
int internalpos
Definition: elog.h:395
#define PG_DIAG_SEVERITY
Definition: postgres_ext.h:55
MemoryContext CurrentMemoryContext
Definition: mcxt.c:42
static bool pq_mq_busy
Definition: pqmq.c:25
#define PG_DIAG_STATEMENT_POSITION
Definition: postgres_ext.h:61
#define WARNING
Definition: elog.h:40
char * datatype_name
Definition: elog.h:392
int elevel
Definition: elog.h:371
char * detail
Definition: elog.h:383
static pid_t pq_mq_parallel_leader_backend_id
Definition: pqmq.c:27
#define InvalidBackendId
Definition: backendid.h:23
uintptr_t Datum
Definition: postgres.h:411
int BackendId
Definition: backendid.h:21
static bool mq_is_send_pending(void)
Definition: pqmq.c:104
#define PG_DIAG_DATATYPE_NAME
Definition: postgres_ext.h:68
#define PG_DIAG_CONSTRAINT_NAME
Definition: postgres_ext.h:69
#define PG_DIAG_SEVERITY_NONLOCALIZED
Definition: postgres_ext.h:56
void pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh)
Definition: pqmq.c:51
#define PG_DIAG_INTERNAL_QUERY
Definition: postgres_ext.h:63
#define PG_DIAG_MESSAGE_HINT
Definition: postgres_ext.h:60
static struct @143 value
char * column_name
Definition: elog.h:391
#define NOTICE
Definition: elog.h:37
int pq_getmsgbyte(StringInfo msg)
Definition: pqformat.c:401
shm_mq_result
Definition: shm_mq.h:36
#define Assert(condition)
Definition: c.h:804
char * hint
Definition: elog.h:385
static pid_t pq_mq_parallel_leader_pid
Definition: pqmq.c:26
char * context
Definition: elog.h:386
struct MemoryContextData * assoc_context
Definition: elog.h:400
int32 pg_strtoint32(const char *s)
Definition: numutils.c:263
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:362
#define elog(elevel,...)
Definition: elog.h:232
void pq_set_parallel_leader(pid_t pid, BackendId backend_id)
Definition: pqmq.c:76
void * arg
static int mq_flush(void)
Definition: pqmq.c:90
struct Latch * MyLatch
Definition: globals.c:57
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:120
void pq_getmsgend(StringInfo msg)
Definition: pqformat.c:637
CommandDest whereToSendOutput
Definition: postgres.c:92
char * constraint_name
Definition: elog.h:393
#define WL_LATCH_SET
Definition: latch.h:125
#define PG_DIAG_SOURCE_FUNCTION
Definition: postgres_ext.h:72
ProtocolVersion FrontendProtocol
Definition: globals.c:28
char * message
Definition: elog.h:382
#define PG_DIAG_CONTEXT
Definition: postgres_ext.h:64
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:130