PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
pqmq.c File Reference
#include "postgres.h"
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "libpq/pqmq.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
Include dependency graph for pqmq.c:

Go to the source code of this file.

Functions

static void pq_cleanup_redirect_to_shm_mq (dsm_segment *seg, Datum arg)
 
static void mq_comm_reset (void)
 
static int mq_flush (void)
 
static int mq_flush_if_writable (void)
 
static bool mq_is_send_pending (void)
 
static int mq_putmessage (char msgtype, const char *s, size_t len)
 
static void mq_putmessage_noblock (char msgtype, const char *s, size_t len)
 
static void mq_startcopyout (void)
 
static void mq_endcopyout (bool errorAbort)
 
void pq_redirect_to_shm_mq (dsm_segment *seg, shm_mq_handle *mqh)
 
void pq_set_parallel_master (pid_t pid, BackendId backend_id)
 
void pq_parse_errornotice (StringInfo msg, ErrorData *edata)
 

Variables

static shm_mqpq_mq
 
static shm_mq_handlepq_mq_handle
 
static bool pq_mq_busy = false
 
static pid_t pq_mq_parallel_master_pid = 0
 
static pid_t pq_mq_parallel_master_backend_id = InvalidBackendId
 
static PQcommMethods PqCommMqMethods
 

Function Documentation

static void mq_comm_reset ( void  )
static

Definition at line 91 of file pqmq.c.

92 {
93  /* Nothing to do. */
94 }
static void mq_endcopyout ( bool  errorAbort)
static

Definition at line 209 of file pqmq.c.

210 {
211  /* Nothing to do. */
212 }
static int mq_flush ( void  )
static

Definition at line 97 of file pqmq.c.

98 {
99  /* Nothing to do. */
100  return 0;
101 }
static int mq_flush_if_writable ( void  )
static

Definition at line 104 of file pqmq.c.

105 {
106  /* Nothing to do. */
107  return 0;
108 }
static bool mq_is_send_pending ( void  )
static

Definition at line 111 of file pqmq.c.

112 {
113  /* There's never anything pending. */
114  return 0;
115 }
static int mq_putmessage ( char  msgtype,
const char *  s,
size_t  len 
)
static

Definition at line 123 of file pqmq.c.

References Assert, CHECK_FOR_INTERRUPTS, shm_mq_iovec::data, shm_mq_iovec::len, MyLatch, NULL, pq_mq_busy, pq_mq_parallel_master_backend_id, pq_mq_parallel_master_pid, PROCSIG_PARALLEL_MESSAGE, ResetLatch(), result, SendProcSignal(), shm_mq_detach(), SHM_MQ_DETACHED, shm_mq_sendv(), SHM_MQ_SUCCESS, SHM_MQ_WOULD_BLOCK, WAIT_EVENT_MQ_PUT_MESSAGE, WaitLatch(), and WL_LATCH_SET.

124 {
125  shm_mq_iovec iov[2];
127 
128  /*
129  * If we're sending a message, and we have to wait because the queue is
130  * full, and then we get interrupted, and that interrupt results in trying
131  * to send another message, we respond by detaching the queue. There's no
132  * way to return to the original context, but even if there were, just
133  * queueing the message would amount to indefinitely postponing the
134  * response to the interrupt. So we do this instead.
135  */
136  if (pq_mq_busy)
137  {
138  if (pq_mq != NULL)
140  pq_mq = NULL;
141  pq_mq_handle = NULL;
142  return EOF;
143  }
144 
145  /*
146  * If the message queue is already gone, just ignore the message. This
147  * doesn't necessarily indicate a problem; for example, DEBUG messages can
148  * be generated late in the shutdown sequence, after all DSMs have already
149  * been detached.
150  */
151  if (pq_mq == NULL)
152  return 0;
153 
154  pq_mq_busy = true;
155 
156  iov[0].data = &msgtype;
157  iov[0].len = 1;
158  iov[1].data = s;
159  iov[1].len = len;
160 
162 
163  for (;;)
164  {
165  result = shm_mq_sendv(pq_mq_handle, iov, 2, true);
166 
167  if (pq_mq_parallel_master_pid != 0)
171 
172  if (result != SHM_MQ_WOULD_BLOCK)
173  break;
174 
179  }
180 
181  pq_mq_busy = false;
182 
183  Assert(result == SHM_MQ_SUCCESS || result == SHM_MQ_DETACHED);
184  if (result != SHM_MQ_SUCCESS)
185  return EOF;
186  return 0;
187 }
static shm_mq_handle * pq_mq_handle
Definition: pqmq.c:25
return result
Definition: formatting.c:1633
void ResetLatch(volatile Latch *latch)
Definition: latch.c:497
Size len
Definition: shm_mq.h:32
void shm_mq_detach(shm_mq *mq)
Definition: shm_mq.c:778
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
const char * data
Definition: shm_mq.h:31
static shm_mq * pq_mq
Definition: pqmq.c:24
static bool pq_mq_busy
Definition: pqmq.c:26
static pid_t pq_mq_parallel_master_pid
Definition: pqmq.c:27
static pid_t pq_mq_parallel_master_backend_id
Definition: pqmq.c:28
shm_mq_result
Definition: shm_mq.h:36
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:676
shm_mq_result shm_mq_sendv(shm_mq_handle *mqh, shm_mq_iovec *iov, int iovcnt, bool nowait)
Definition: shm_mq.c:347
struct Latch * MyLatch
Definition: globals.c:52
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
#define WL_LATCH_SET
Definition: latch.h:124
static void mq_putmessage_noblock ( char  msgtype,
const char *  s,
size_t  len 
)
static

Definition at line 190 of file pqmq.c.

References elog, and ERROR.

191 {
192  /*
193  * While the shm_mq machinery does support sending a message in
194  * non-blocking mode, there's currently no way to try sending beginning to
195  * send the message that doesn't also commit us to completing the
196  * transmission. This could be improved in the future, but for now we
197  * don't need it.
198  */
199  elog(ERROR, "not currently supported");
200 }
#define ERROR
Definition: elog.h:43
#define elog
Definition: elog.h:219
static void mq_startcopyout ( void  )
static

Definition at line 203 of file pqmq.c.

204 {
205  /* Nothing to do. */
206 }
static void pq_cleanup_redirect_to_shm_mq ( dsm_segment seg,
Datum  arg 
)
static

Definition at line 71 of file pqmq.c.

References DestNone, NULL, and whereToSendOutput.

Referenced by pq_redirect_to_shm_mq().

72 {
73  pq_mq = NULL;
76 }
static shm_mq_handle * pq_mq_handle
Definition: pqmq.c:25
Definition: dest.h:88
static shm_mq * pq_mq
Definition: pqmq.c:24
#define NULL
Definition: c.h:229
CommandDest whereToSendOutput
Definition: postgres.c:88
void pq_parse_errornotice ( StringInfo  msg,
ErrorData edata 
)

Definition at line 219 of file pqmq.c.

References ErrorData::assoc_context, ErrorData::column_name, ErrorData::constraint_name, ErrorData::context, CurrentMemoryContext, ErrorData::cursorpos, ErrorData::datatype_name, DEBUG1, ErrorData::detail, ErrorData::elevel, elog, ERROR, FATAL, ErrorData::filename, ErrorData::funcname, ErrorData::hint, INFO, ErrorData::internalpos, ErrorData::internalquery, ErrorData::lineno, LOG, MAKE_SQLSTATE, MemSet, ErrorData::message, NOTICE, PANIC, pg_atoi(), PG_DIAG_COLUMN_NAME, PG_DIAG_CONSTRAINT_NAME, PG_DIAG_CONTEXT, PG_DIAG_DATATYPE_NAME, PG_DIAG_INTERNAL_POSITION, PG_DIAG_INTERNAL_QUERY, PG_DIAG_MESSAGE_DETAIL, PG_DIAG_MESSAGE_HINT, PG_DIAG_MESSAGE_PRIMARY, PG_DIAG_SCHEMA_NAME, PG_DIAG_SEVERITY, PG_DIAG_SEVERITY_NONLOCALIZED, PG_DIAG_SOURCE_FILE, PG_DIAG_SOURCE_FUNCTION, PG_DIAG_SOURCE_LINE, PG_DIAG_SQLSTATE, PG_DIAG_STATEMENT_POSITION, PG_DIAG_TABLE_NAME, pq_getmsgbyte(), pq_getmsgend(), pq_getmsgrawstring(), pstrdup(), ErrorData::schema_name, ErrorData::sqlerrcode, ErrorData::table_name, value, and WARNING.

Referenced by HandleParallelMessage().

220 {
221  /* Initialize edata with reasonable defaults. */
222  MemSet(edata, 0, sizeof(ErrorData));
223  edata->elevel = ERROR;
225 
226  /* Loop over fields and extract each one. */
227  for (;;)
228  {
229  char code = pq_getmsgbyte(msg);
230  const char *value;
231 
232  if (code == '\0')
233  {
234  pq_getmsgend(msg);
235  break;
236  }
237  value = pq_getmsgrawstring(msg);
238 
239  switch (code)
240  {
241  case PG_DIAG_SEVERITY:
242  /* ignore, trusting we'll get a nonlocalized version */
243  break;
245  if (strcmp(value, "DEBUG") == 0)
246  {
247  /*
248  * We can't reconstruct the exact DEBUG level, but
249  * presumably it was >= client_min_messages, so select
250  * DEBUG1 to ensure we'll pass it on to the client.
251  */
252  edata->elevel = DEBUG1;
253  }
254  else if (strcmp(value, "LOG") == 0)
255  {
256  /*
257  * It can't be LOG_SERVER_ONLY, or the worker wouldn't
258  * have sent it to us; so LOG is the correct value.
259  */
260  edata->elevel = LOG;
261  }
262  else if (strcmp(value, "INFO") == 0)
263  edata->elevel = INFO;
264  else if (strcmp(value, "NOTICE") == 0)
265  edata->elevel = NOTICE;
266  else if (strcmp(value, "WARNING") == 0)
267  edata->elevel = WARNING;
268  else if (strcmp(value, "ERROR") == 0)
269  edata->elevel = ERROR;
270  else if (strcmp(value, "FATAL") == 0)
271  edata->elevel = FATAL;
272  else if (strcmp(value, "PANIC") == 0)
273  edata->elevel = PANIC;
274  else
275  elog(ERROR, "unrecognized error severity: \"%s\"", value);
276  break;
277  case PG_DIAG_SQLSTATE:
278  if (strlen(value) != 5)
279  elog(ERROR, "invalid SQLSTATE: \"%s\"", value);
280  edata->sqlerrcode = MAKE_SQLSTATE(value[0], value[1], value[2],
281  value[3], value[4]);
282  break;
284  edata->message = pstrdup(value);
285  break;
287  edata->detail = pstrdup(value);
288  break;
290  edata->hint = pstrdup(value);
291  break;
293  edata->cursorpos = pg_atoi(value, sizeof(int), '\0');
294  break;
296  edata->internalpos = pg_atoi(value, sizeof(int), '\0');
297  break;
299  edata->internalquery = pstrdup(value);
300  break;
301  case PG_DIAG_CONTEXT:
302  edata->context = pstrdup(value);
303  break;
304  case PG_DIAG_SCHEMA_NAME:
305  edata->schema_name = pstrdup(value);
306  break;
307  case PG_DIAG_TABLE_NAME:
308  edata->table_name = pstrdup(value);
309  break;
310  case PG_DIAG_COLUMN_NAME:
311  edata->column_name = pstrdup(value);
312  break;
314  edata->datatype_name = pstrdup(value);
315  break;
317  edata->constraint_name = pstrdup(value);
318  break;
319  case PG_DIAG_SOURCE_FILE:
320  edata->filename = pstrdup(value);
321  break;
322  case PG_DIAG_SOURCE_LINE:
323  edata->lineno = pg_atoi(value, sizeof(int), '\0');
324  break;
326  edata->funcname = pstrdup(value);
327  break;
328  default:
329  elog(ERROR, "unrecognized error field code: %d", (int) code);
330  break;
331  }
332  }
333 }
char * schema_name
Definition: elog.h:349
#define DEBUG1
Definition: elog.h:25
#define PG_DIAG_MESSAGE_PRIMARY
Definition: postgres_ext.h:58
#define PG_DIAG_SCHEMA_NAME
Definition: postgres_ext.h:65
int sqlerrcode
Definition: elog.h:342
#define PG_DIAG_MESSAGE_DETAIL
Definition: postgres_ext.h:59
#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
char * pstrdup(const char *in)
Definition: mcxt.c:1077
#define PG_DIAG_TABLE_NAME
Definition: postgres_ext.h:66
int lineno
Definition: elog.h:338
#define PG_DIAG_SOURCE_LINE
Definition: postgres_ext.h:71
#define MemSet(start, val, len)
Definition: c.h:858
#define INFO
Definition: elog.h:33
char * internalquery
Definition: elog.h:356
#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:650
#define PG_DIAG_SQLSTATE
Definition: postgres_ext.h:57
#define PANIC
Definition: elog.h:53
int cursorpos
Definition: elog.h:354
#define PG_DIAG_SOURCE_FILE
Definition: postgres_ext.h:70
#define ERROR
Definition: elog.h:43
const char * filename
Definition: elog.h:337
#define FATAL
Definition: elog.h:52
static struct @121 value
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
#define PG_DIAG_STATEMENT_POSITION
Definition: postgres_ext.h:61
#define WARNING
Definition: elog.h:40
char * datatype_name
Definition: elog.h:352
int elevel
Definition: elog.h:331
char * detail
Definition: elog.h:344
#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
#define PG_DIAG_INTERNAL_QUERY
Definition: postgres_ext.h:63
#define PG_DIAG_MESSAGE_HINT
Definition: postgres_ext.h:60
char * column_name
Definition: elog.h:351
#define NOTICE
Definition: elog.h:37
int pq_getmsgbyte(StringInfo msg)
Definition: pqformat.c:432
char * hint
Definition: elog.h:346
char * context
Definition: elog.h:347
struct MemoryContextData * assoc_context
Definition: elog.h:360
#define elog
Definition: elog.h:219
void pq_getmsgend(StringInfo msg)
Definition: pqformat.c:677
char * constraint_name
Definition: elog.h:353
#define PG_DIAG_SOURCE_FUNCTION
Definition: postgres_ext.h:72
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
void pq_redirect_to_shm_mq ( dsm_segment seg,
shm_mq_handle mqh 
)

Definition at line 56 of file pqmq.c.

References DestRemote, FrontendProtocol, on_dsm_detach(), PG_PROTOCOL_LATEST, pq_cleanup_redirect_to_shm_mq(), PqCommMethods, PqCommMqMethods, shm_mq_get_queue(), and whereToSendOutput.

Referenced by ParallelWorkerMain().

57 {
59  pq_mq = shm_mq_get_queue(mqh);
60  pq_mq_handle = mqh;
64 }
static shm_mq_handle * pq_mq_handle
Definition: pqmq.c:25
PQcommMethods * PqCommMethods
Definition: pqcomm.c:184
#define PG_PROTOCOL_LATEST
Definition: pqcomm.h:111
static void pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg)
Definition: pqmq.c:71
void on_dsm_detach(dsm_segment *seg, on_dsm_detach_callback function, Datum arg)
Definition: dsm.c:1037
static PQcommMethods PqCommMqMethods
Definition: pqmq.c:40
static shm_mq * pq_mq
Definition: pqmq.c:24
uintptr_t Datum
Definition: postgres.h:372
shm_mq * shm_mq_get_queue(shm_mq_handle *mqh)
Definition: shm_mq.c:802
CommandDest whereToSendOutput
Definition: postgres.c:88
ProtocolVersion FrontendProtocol
Definition: globals.c:27
void pq_set_parallel_master ( pid_t  pid,
BackendId  backend_id 
)

Definition at line 83 of file pqmq.c.

References Assert, pq_mq_parallel_master_backend_id, pq_mq_parallel_master_pid, and PqCommMethods.

Referenced by ParallelWorkerMain().

84 {
88 }
PQcommMethods * PqCommMethods
Definition: pqcomm.c:184
static PQcommMethods PqCommMqMethods
Definition: pqmq.c:40
static pid_t pq_mq_parallel_master_pid
Definition: pqmq.c:27
static pid_t pq_mq_parallel_master_backend_id
Definition: pqmq.c:28
#define Assert(condition)
Definition: c.h:676

Variable Documentation

shm_mq* pq_mq
static

Definition at line 24 of file pqmq.c.

bool pq_mq_busy = false
static

Definition at line 26 of file pqmq.c.

Referenced by mq_putmessage().

shm_mq_handle* pq_mq_handle
static

Definition at line 25 of file pqmq.c.

pid_t pq_mq_parallel_master_backend_id = InvalidBackendId
static

Definition at line 28 of file pqmq.c.

Referenced by mq_putmessage(), and pq_set_parallel_master().

pid_t pq_mq_parallel_master_pid = 0
static

Definition at line 27 of file pqmq.c.

Referenced by mq_putmessage(), and pq_set_parallel_master().

PQcommMethods PqCommMqMethods
static
Initial value:
= {
}
static int mq_putmessage(char msgtype, const char *s, size_t len)
Definition: pqmq.c:123
static int mq_flush_if_writable(void)
Definition: pqmq.c:104
static void mq_putmessage_noblock(char msgtype, const char *s, size_t len)
Definition: pqmq.c:190
static void mq_comm_reset(void)
Definition: pqmq.c:91
static void mq_startcopyout(void)
Definition: pqmq.c:203
static bool mq_is_send_pending(void)
Definition: pqmq.c:111
static void mq_endcopyout(bool errorAbort)
Definition: pqmq.c:209
static int mq_flush(void)
Definition: pqmq.c:97

Definition at line 40 of file pqmq.c.

Referenced by pq_redirect_to_shm_mq().