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-2017, 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_master_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 static void mq_startcopyout(void);
37 static void mq_endcopyout(bool errorAbort);
38 
41  mq_flush,
48 };
49 
50 /*
51  * Arrange to redirect frontend/backend protocol messages to a shared-memory
52  * message queue.
53  */
54 void
56 {
58  pq_mq_handle = mqh;
62 }
63 
64 /*
65  * When the DSM that contains our shm_mq goes away, we need to stop sending
66  * messages to it.
67  */
68 static void
70 {
71  pq_mq_handle = NULL;
73 }
74 
75 /*
76  * Arrange to SendProcSignal() to the parallel master each time we transmit
77  * message data via the shm_mq.
78  */
79 void
80 pq_set_parallel_master(pid_t pid, BackendId backend_id)
81 {
82  Assert(PqCommMethods == &PqCommMqMethods);
85 }
86 
87 static void
89 {
90  /* Nothing to do. */
91 }
92 
93 static int
94 mq_flush(void)
95 {
96  /* Nothing to do. */
97  return 0;
98 }
99 
100 static int
102 {
103  /* Nothing to do. */
104  return 0;
105 }
106 
107 static bool
109 {
110  /* There's never anything pending. */
111  return 0;
112 }
113 
114 /*
115  * Transmit a libpq protocol message to the shared memory message queue
116  * selected via pq_mq_handle. We don't include a length word, because the
117  * receiver will know the length of the message from shm_mq_receive().
118  */
119 static int
120 mq_putmessage(char msgtype, const char *s, size_t len)
121 {
122  shm_mq_iovec iov[2];
123  shm_mq_result result;
124 
125  /*
126  * If we're sending a message, and we have to wait because the queue is
127  * full, and then we get interrupted, and that interrupt results in trying
128  * to send another message, we respond by detaching the queue. There's no
129  * way to return to the original context, but even if there were, just
130  * queueing the message would amount to indefinitely postponing the
131  * response to the interrupt. So we do this instead.
132  */
133  if (pq_mq_busy)
134  {
135  if (pq_mq_handle != NULL)
136  shm_mq_detach(pq_mq_handle);
137  pq_mq_handle = NULL;
138  return EOF;
139  }
140 
141  /*
142  * If the message queue is already gone, just ignore the message. This
143  * doesn't necessarily indicate a problem; for example, DEBUG messages can
144  * be generated late in the shutdown sequence, after all DSMs have already
145  * been detached.
146  */
147  if (pq_mq_handle == NULL)
148  return 0;
149 
150  pq_mq_busy = true;
151 
152  iov[0].data = &msgtype;
153  iov[0].len = 1;
154  iov[1].data = s;
155  iov[1].len = len;
156 
157  Assert(pq_mq_handle != NULL);
158 
159  for (;;)
160  {
161  result = shm_mq_sendv(pq_mq_handle, iov, 2, true);
162 
163  if (pq_mq_parallel_master_pid != 0)
167 
168  if (result != SHM_MQ_WOULD_BLOCK)
169  break;
170 
175  }
176 
177  pq_mq_busy = false;
178 
179  Assert(result == SHM_MQ_SUCCESS || result == SHM_MQ_DETACHED);
180  if (result != SHM_MQ_SUCCESS)
181  return EOF;
182  return 0;
183 }
184 
185 static void
186 mq_putmessage_noblock(char msgtype, const char *s, size_t len)
187 {
188  /*
189  * While the shm_mq machinery does support sending a message in
190  * non-blocking mode, there's currently no way to try sending beginning to
191  * send the message that doesn't also commit us to completing the
192  * transmission. This could be improved in the future, but for now we
193  * don't need it.
194  */
195  elog(ERROR, "not currently supported");
196 }
197 
198 static void
200 {
201  /* Nothing to do. */
202 }
203 
204 static void
205 mq_endcopyout(bool errorAbort)
206 {
207  /* Nothing to do. */
208 }
209 
210 /*
211  * Parse an ErrorResponse or NoticeResponse payload and populate an ErrorData
212  * structure with the results.
213  */
214 void
216 {
217  /* Initialize edata with reasonable defaults. */
218  MemSet(edata, 0, sizeof(ErrorData));
219  edata->elevel = ERROR;
221 
222  /* Loop over fields and extract each one. */
223  for (;;)
224  {
225  char code = pq_getmsgbyte(msg);
226  const char *value;
227 
228  if (code == '\0')
229  {
230  pq_getmsgend(msg);
231  break;
232  }
233  value = pq_getmsgrawstring(msg);
234 
235  switch (code)
236  {
237  case PG_DIAG_SEVERITY:
238  /* ignore, trusting we'll get a nonlocalized version */
239  break;
241  if (strcmp(value, "DEBUG") == 0)
242  {
243  /*
244  * We can't reconstruct the exact DEBUG level, but
245  * presumably it was >= client_min_messages, so select
246  * DEBUG1 to ensure we'll pass it on to the client.
247  */
248  edata->elevel = DEBUG1;
249  }
250  else if (strcmp(value, "LOG") == 0)
251  {
252  /*
253  * It can't be LOG_SERVER_ONLY, or the worker wouldn't
254  * have sent it to us; so LOG is the correct value.
255  */
256  edata->elevel = LOG;
257  }
258  else if (strcmp(value, "INFO") == 0)
259  edata->elevel = INFO;
260  else if (strcmp(value, "NOTICE") == 0)
261  edata->elevel = NOTICE;
262  else if (strcmp(value, "WARNING") == 0)
263  edata->elevel = WARNING;
264  else if (strcmp(value, "ERROR") == 0)
265  edata->elevel = ERROR;
266  else if (strcmp(value, "FATAL") == 0)
267  edata->elevel = FATAL;
268  else if (strcmp(value, "PANIC") == 0)
269  edata->elevel = PANIC;
270  else
271  elog(ERROR, "unrecognized error severity: \"%s\"", value);
272  break;
273  case PG_DIAG_SQLSTATE:
274  if (strlen(value) != 5)
275  elog(ERROR, "invalid SQLSTATE: \"%s\"", value);
276  edata->sqlerrcode = MAKE_SQLSTATE(value[0], value[1], value[2],
277  value[3], value[4]);
278  break;
280  edata->message = pstrdup(value);
281  break;
283  edata->detail = pstrdup(value);
284  break;
286  edata->hint = pstrdup(value);
287  break;
289  edata->cursorpos = pg_atoi(value, sizeof(int), '\0');
290  break;
292  edata->internalpos = pg_atoi(value, sizeof(int), '\0');
293  break;
295  edata->internalquery = pstrdup(value);
296  break;
297  case PG_DIAG_CONTEXT:
298  edata->context = pstrdup(value);
299  break;
300  case PG_DIAG_SCHEMA_NAME:
301  edata->schema_name = pstrdup(value);
302  break;
303  case PG_DIAG_TABLE_NAME:
304  edata->table_name = pstrdup(value);
305  break;
306  case PG_DIAG_COLUMN_NAME:
307  edata->column_name = pstrdup(value);
308  break;
310  edata->datatype_name = pstrdup(value);
311  break;
313  edata->constraint_name = pstrdup(value);
314  break;
315  case PG_DIAG_SOURCE_FILE:
316  edata->filename = pstrdup(value);
317  break;
318  case PG_DIAG_SOURCE_LINE:
319  edata->lineno = pg_atoi(value, sizeof(int), '\0');
320  break;
322  edata->funcname = pstrdup(value);
323  break;
324  default:
325  elog(ERROR, "unrecognized error field code: %d", (int) code);
326  break;
327  }
328  }
329 }
char * schema_name
Definition: elog.h:349
#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
PQcommMethods * PqCommMethods
Definition: pqcomm.c:184
static int mq_putmessage(char msgtype, const char *s, size_t len)
Definition: pqmq.c:120
int sqlerrcode
Definition: elog.h:342
#define PG_DIAG_MESSAGE_DETAIL
Definition: postgres_ext.h:59
void shm_mq_detach(shm_mq_handle *mqh)
Definition: shm_mq.c:775
#define PG_DIAG_COLUMN_NAME
Definition: postgres_ext.h:67
#define MAKE_SQLSTATE(ch1, ch2, ch3, ch4, ch5)
Definition: elog.h:62
const char * funcname
Definition: elog.h:339
static int mq_flush_if_writable(void)
Definition: pqmq.c:101
#define PG_PROTOCOL_LATEST
Definition: pqcomm.h:111
char * pstrdup(const char *in)
Definition: mcxt.c:1076
#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:69
int lineno
Definition: elog.h:338
#define PG_DIAG_SOURCE_LINE
Definition: postgres_ext.h:71
#define MemSet(start, val, len)
Definition: c.h:853
#define INFO
Definition: elog.h:33
char * internalquery
Definition: elog.h:356
void ResetLatch(volatile Latch *latch)
Definition: latch.c:497
#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:1037
#define PG_DIAG_SQLSTATE
Definition: postgres_ext.h:57
#define PANIC
Definition: elog.h:53
static PQcommMethods PqCommMqMethods
Definition: pqmq.c:39
Size len
Definition: shm_mq.h:32
static void mq_putmessage_noblock(char msgtype, const char *s, size_t len)
Definition: pqmq.c:186
int cursorpos
Definition: elog.h:354
int SendProcSignal(pid_t pid, ProcSignalReason reason, BackendId backendId)
Definition: procsignal.c:180
int WaitLatch(volatile Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:336
#define PG_DIAG_SOURCE_FILE
Definition: postgres_ext.h:70
const char * data
Definition: shm_mq.h:31
#define ERROR
Definition: elog.h:43
const char * filename
Definition: elog.h:337
static void mq_comm_reset(void)
Definition: pqmq.c:88
void pq_parse_errornotice(StringInfo msg, ErrorData *edata)
Definition: pqmq.c:215
#define FATAL
Definition: elog.h:52
static struct @121 value
Definition: dest.h:88
char * table_name
Definition: elog.h:350
int internalpos
Definition: elog.h:355
#define PG_DIAG_SEVERITY
Definition: postgres_ext.h:55
MemoryContext CurrentMemoryContext
Definition: mcxt.c:37
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
static void mq_startcopyout(void)
Definition: pqmq.c:199
char * datatype_name
Definition: elog.h:352
int elevel
Definition: elog.h:331
char * detail
Definition: elog.h:344
#define InvalidBackendId
Definition: backendid.h:23
uintptr_t Datum
Definition: postgres.h:372
int BackendId
Definition: backendid.h:21
static bool mq_is_send_pending(void)
Definition: pqmq.c:108
#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:55
static pid_t pq_mq_parallel_master_pid
Definition: pqmq.c:26
#define PG_DIAG_INTERNAL_QUERY
Definition: postgres_ext.h:63
#define PG_DIAG_MESSAGE_HINT
Definition: postgres_ext.h:60
static pid_t pq_mq_parallel_master_backend_id
Definition: pqmq.c:27
char * column_name
Definition: elog.h:351
#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:670
char * hint
Definition: elog.h:346
shm_mq_result shm_mq_sendv(shm_mq_handle *mqh, shm_mq_iovec *iov, int iovcnt, bool nowait)
Definition: shm_mq.c:351
char * context
Definition: elog.h:347
struct MemoryContextData * assoc_context
Definition: elog.h:360
void pq_set_parallel_master(pid_t pid, BackendId backend_id)
Definition: pqmq.c:80
static void mq_endcopyout(bool errorAbort)
Definition: pqmq.c:205
void * arg
static int mq_flush(void)
Definition: pqmq.c:94
struct Latch * MyLatch
Definition: globals.c:52
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
#define elog
Definition: elog.h:219
void pq_getmsgend(StringInfo msg)
Definition: pqformat.c:637
CommandDest whereToSendOutput
Definition: postgres.c:88
char * constraint_name
Definition: elog.h:353
#define WL_LATCH_SET
Definition: latch.h:124
#define PG_DIAG_SOURCE_FUNCTION
Definition: postgres_ext.h:72
ProtocolVersion FrontendProtocol
Definition: globals.c:27
char * message
Definition: elog.h:343
int32 pg_atoi(const char *s, int size, int c)
Definition: numutils.c:37
#define PG_DIAG_CONTEXT
Definition: postgres_ext.h:64