PostgreSQL Source Code  git master
pqmq.c File Reference
#include "postgres.h"
#include "access/parallel.h"
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "libpq/pqmq.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "replication/logicalworker.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)
 
void pq_redirect_to_shm_mq (dsm_segment *seg, shm_mq_handle *mqh)
 
void pq_set_parallel_leader (pid_t pid, ProcNumber procNumber)
 
void pq_parse_errornotice (StringInfo msg, ErrorData *edata)
 

Variables

static shm_mq_handlepq_mq_handle
 
static bool pq_mq_busy = false
 
static pid_t pq_mq_parallel_leader_pid = 0
 
static pid_t pq_mq_parallel_leader_proc_number = INVALID_PROC_NUMBER
 
static const PQcommMethods PqCommMqMethods
 

Function Documentation

◆ mq_comm_reset()

static void mq_comm_reset ( void  )
static

Definition at line 86 of file pqmq.c.

87 {
88  /* Nothing to do. */
89 }

◆ mq_flush()

static int mq_flush ( void  )
static

Definition at line 92 of file pqmq.c.

93 {
94  /* Nothing to do. */
95  return 0;
96 }

◆ mq_flush_if_writable()

static int mq_flush_if_writable ( void  )
static

Definition at line 99 of file pqmq.c.

100 {
101  /* Nothing to do. */
102  return 0;
103 }

◆ mq_is_send_pending()

static bool mq_is_send_pending ( void  )
static

Definition at line 106 of file pqmq.c.

107 {
108  /* There's never anything pending. */
109  return 0;
110 }

◆ mq_putmessage()

static int mq_putmessage ( char  msgtype,
const char *  s,
size_t  len 
)
static

Definition at line 118 of file pqmq.c.

119 {
120  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  */
131  if (pq_mq_busy)
132  {
133  if (pq_mq_handle != NULL)
135  pq_mq_handle = NULL;
136  return EOF;
137  }
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  */
145  if (pq_mq_handle == NULL)
146  return 0;
147 
148  pq_mq_busy = true;
149 
150  iov[0].data = &msgtype;
151  iov[0].len = 1;
152  iov[1].data = s;
153  iov[1].len = len;
154 
155  Assert(pq_mq_handle != NULL);
156 
157  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  */
164  result = shm_mq_sendv(pq_mq_handle, iov, 2, true, true);
165 
166  if (pq_mq_parallel_leader_pid != 0)
167  {
172  else
173  {
178  }
179  }
180 
181  if (result != SHM_MQ_WOULD_BLOCK)
182  break;
183 
185  WAIT_EVENT_MESSAGE_QUEUE_PUT_MESSAGE);
188  }
189 
190  pq_mq_busy = false;
191 
192  Assert(result == SHM_MQ_SUCCESS || result == SHM_MQ_DETACHED);
193  if (result != SHM_MQ_SUCCESS)
194  return EOF;
195  return 0;
196 }
bool IsLogicalParallelApplyWorker(void)
Definition: worker.c:4754
#define Assert(condition)
Definition: c.h:858
struct Latch * MyLatch
Definition: globals.c:61
#define IsParallelWorker()
Definition: parallel.h:60
void ResetLatch(Latch *latch)
Definition: latch.c:724
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:517
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:132
#define WL_LATCH_SET
Definition: latch.h:127
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
const void size_t len
static bool pq_mq_busy
Definition: pqmq.c:27
static pid_t pq_mq_parallel_leader_proc_number
Definition: pqmq.c:29
static pid_t pq_mq_parallel_leader_pid
Definition: pqmq.c:28
static shm_mq_handle * pq_mq_handle
Definition: pqmq.c:26
int SendProcSignal(pid_t pid, ProcSignalReason reason, ProcNumber procNumber)
Definition: procsignal.c:257
@ PROCSIG_PARALLEL_MESSAGE
Definition: procsignal.h:34
@ PROCSIG_PARALLEL_APPLY_MESSAGE
Definition: procsignal.h:38
void shm_mq_detach(shm_mq_handle *mqh)
Definition: shm_mq.c:843
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:361
shm_mq_result
Definition: shm_mq.h:37
@ SHM_MQ_SUCCESS
Definition: shm_mq.h:38
@ SHM_MQ_WOULD_BLOCK
Definition: shm_mq.h:39
@ SHM_MQ_DETACHED
Definition: shm_mq.h:40
const char * data
Definition: shm_mq.h:31
Size len
Definition: shm_mq.h:32

References Assert, CHECK_FOR_INTERRUPTS, shm_mq_iovec::data, IsLogicalParallelApplyWorker(), IsParallelWorker, shm_mq_iovec::len, len, MyLatch, pq_mq_busy, pq_mq_handle, pq_mq_parallel_leader_pid, pq_mq_parallel_leader_proc_number, PROCSIG_PARALLEL_APPLY_MESSAGE, PROCSIG_PARALLEL_MESSAGE, ResetLatch(), SendProcSignal(), shm_mq_detach(), SHM_MQ_DETACHED, shm_mq_sendv(), SHM_MQ_SUCCESS, SHM_MQ_WOULD_BLOCK, WaitLatch(), WL_EXIT_ON_PM_DEATH, and WL_LATCH_SET.

◆ mq_putmessage_noblock()

static void mq_putmessage_noblock ( char  msgtype,
const char *  s,
size_t  len 
)
static

Definition at line 199 of file pqmq.c.

200 {
201  /*
202  * 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
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.
207  */
208  elog(ERROR, "not currently supported");
209 }
#define ERROR
Definition: elog.h:39
#define elog(elevel,...)
Definition: elog.h:224

References elog, and ERROR.

◆ pq_cleanup_redirect_to_shm_mq()

static void pq_cleanup_redirect_to_shm_mq ( dsm_segment seg,
Datum  arg 
)
static

Definition at line 67 of file pqmq.c.

68 {
69  pq_mq_handle = NULL;
71 }
@ DestNone
Definition: dest.h:87
CommandDest whereToSendOutput
Definition: postgres.c:90

References DestNone, pq_mq_handle, and whereToSendOutput.

Referenced by pq_redirect_to_shm_mq().

◆ pq_parse_errornotice()

void pq_parse_errornotice ( StringInfo  msg,
ErrorData edata 
)

Definition at line 216 of file pqmq.c.

217 {
218  /* Initialize edata with reasonable defaults. */
219  MemSet(edata, 0, sizeof(ErrorData));
220  edata->elevel = ERROR;
222 
223  /* Loop over fields and extract each one. */
224  for (;;)
225  {
226  char code = pq_getmsgbyte(msg);
227  const char *value;
228 
229  if (code == '\0')
230  {
231  pq_getmsgend(msg);
232  break;
233  }
234  value = pq_getmsgrawstring(msg);
235 
236  switch (code)
237  {
238  case PG_DIAG_SEVERITY:
239  /* ignore, trusting we'll get a nonlocalized version */
240  break;
242  if (strcmp(value, "DEBUG") == 0)
243  {
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  */
249  edata->elevel = DEBUG1;
250  }
251  else if (strcmp(value, "LOG") == 0)
252  {
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  */
257  edata->elevel = LOG;
258  }
259  else if (strcmp(value, "INFO") == 0)
260  edata->elevel = INFO;
261  else if (strcmp(value, "NOTICE") == 0)
262  edata->elevel = NOTICE;
263  else if (strcmp(value, "WARNING") == 0)
264  edata->elevel = WARNING;
265  else if (strcmp(value, "ERROR") == 0)
266  edata->elevel = ERROR;
267  else if (strcmp(value, "FATAL") == 0)
268  edata->elevel = FATAL;
269  else if (strcmp(value, "PANIC") == 0)
270  edata->elevel = PANIC;
271  else
272  elog(ERROR, "unrecognized error severity: \"%s\"", value);
273  break;
274  case PG_DIAG_SQLSTATE:
275  if (strlen(value) != 5)
276  elog(ERROR, "invalid SQLSTATE: \"%s\"", value);
277  edata->sqlerrcode = MAKE_SQLSTATE(value[0], value[1], value[2],
278  value[3], value[4]);
279  break;
281  edata->message = pstrdup(value);
282  break;
284  edata->detail = pstrdup(value);
285  break;
287  edata->hint = pstrdup(value);
288  break;
290  edata->cursorpos = pg_strtoint32(value);
291  break;
293  edata->internalpos = pg_strtoint32(value);
294  break;
296  edata->internalquery = pstrdup(value);
297  break;
298  case PG_DIAG_CONTEXT:
299  edata->context = pstrdup(value);
300  break;
301  case PG_DIAG_SCHEMA_NAME:
302  edata->schema_name = pstrdup(value);
303  break;
304  case PG_DIAG_TABLE_NAME:
305  edata->table_name = pstrdup(value);
306  break;
307  case PG_DIAG_COLUMN_NAME:
308  edata->column_name = pstrdup(value);
309  break;
311  edata->datatype_name = pstrdup(value);
312  break;
314  edata->constraint_name = pstrdup(value);
315  break;
316  case PG_DIAG_SOURCE_FILE:
317  edata->filename = pstrdup(value);
318  break;
319  case PG_DIAG_SOURCE_LINE:
320  edata->lineno = pg_strtoint32(value);
321  break;
323  edata->funcname = pstrdup(value);
324  break;
325  default:
326  elog(ERROR, "unrecognized error field code: %d", (int) code);
327  break;
328  }
329  }
330 }
#define MemSet(start, val, len)
Definition: c.h:1020
#define LOG
Definition: elog.h:31
#define FATAL
Definition: elog.h:41
#define WARNING
Definition: elog.h:36
#define PANIC
Definition: elog.h:42
#define DEBUG1
Definition: elog.h:30
#define MAKE_SQLSTATE(ch1, ch2, ch3, ch4, ch5)
Definition: elog.h:56
#define NOTICE
Definition: elog.h:35
#define INFO
Definition: elog.h:34
static struct @155 value
char * pstrdup(const char *in)
Definition: mcxt.c:1696
MemoryContext CurrentMemoryContext
Definition: mcxt.c:143
int32 pg_strtoint32(const char *s)
Definition: numutils.c:383
#define PG_DIAG_INTERNAL_QUERY
Definition: postgres_ext.h:62
#define PG_DIAG_SCHEMA_NAME
Definition: postgres_ext.h:64
#define PG_DIAG_CONSTRAINT_NAME
Definition: postgres_ext.h:68
#define PG_DIAG_DATATYPE_NAME
Definition: postgres_ext.h:67
#define PG_DIAG_SOURCE_LINE
Definition: postgres_ext.h:70
#define PG_DIAG_STATEMENT_POSITION
Definition: postgres_ext.h:60
#define PG_DIAG_SOURCE_FILE
Definition: postgres_ext.h:69
#define PG_DIAG_MESSAGE_HINT
Definition: postgres_ext.h:59
#define PG_DIAG_SQLSTATE
Definition: postgres_ext.h:56
#define PG_DIAG_SEVERITY_NONLOCALIZED
Definition: postgres_ext.h:55
#define PG_DIAG_TABLE_NAME
Definition: postgres_ext.h:65
#define PG_DIAG_MESSAGE_PRIMARY
Definition: postgres_ext.h:57
#define PG_DIAG_COLUMN_NAME
Definition: postgres_ext.h:66
#define PG_DIAG_MESSAGE_DETAIL
Definition: postgres_ext.h:58
#define PG_DIAG_CONTEXT
Definition: postgres_ext.h:63
#define PG_DIAG_SEVERITY
Definition: postgres_ext.h:54
#define PG_DIAG_SOURCE_FUNCTION
Definition: postgres_ext.h:71
#define PG_DIAG_INTERNAL_POSITION
Definition: postgres_ext.h:61
void pq_getmsgend(StringInfo msg)
Definition: pqformat.c:635
int pq_getmsgbyte(StringInfo msg)
Definition: pqformat.c:399
const char * pq_getmsgrawstring(StringInfo msg)
Definition: pqformat.c:608
int internalpos
Definition: elog.h:452
char * schema_name
Definition: elog.h:446
char * context
Definition: elog.h:443
char * internalquery
Definition: elog.h:453
int sqlerrcode
Definition: elog.h:438
struct MemoryContextData * assoc_context
Definition: elog.h:457
const char * filename
Definition: elog.h:433
int elevel
Definition: elog.h:428
char * datatype_name
Definition: elog.h:449
char * detail
Definition: elog.h:440
const char * funcname
Definition: elog.h:435
char * table_name
Definition: elog.h:447
char * message
Definition: elog.h:439
int lineno
Definition: elog.h:434
char * hint
Definition: elog.h:442
char * constraint_name
Definition: elog.h:450
int cursorpos
Definition: elog.h:451
char * column_name
Definition: elog.h:448

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_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, pg_strtoint32(), pq_getmsgbyte(), pq_getmsgend(), pq_getmsgrawstring(), pstrdup(), ErrorData::schema_name, ErrorData::sqlerrcode, ErrorData::table_name, value, and WARNING.

Referenced by HandleParallelApplyMessage(), and HandleParallelMessage().

◆ pq_redirect_to_shm_mq()

void pq_redirect_to_shm_mq ( dsm_segment seg,
shm_mq_handle mqh 
)

Definition at line 53 of file pqmq.c.

54 {
56  pq_mq_handle = mqh;
60 }
@ DestRemote
Definition: dest.h:89
void on_dsm_detach(dsm_segment *seg, on_dsm_detach_callback function, Datum arg)
Definition: dsm.c:1132
ProtocolVersion FrontendProtocol
Definition: globals.c:29
uintptr_t Datum
Definition: postgres.h:64
const PQcommMethods * PqCommMethods
Definition: pqcomm.c:164
#define PG_PROTOCOL_LATEST
Definition: pqcomm.h:97
static void pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg)
Definition: pqmq.c:67
static const PQcommMethods PqCommMqMethods
Definition: pqmq.c:39

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

Referenced by ParallelApplyWorkerMain(), and ParallelWorkerMain().

◆ pq_set_parallel_leader()

void pq_set_parallel_leader ( pid_t  pid,
ProcNumber  procNumber 
)

Variable Documentation

◆ pq_mq_busy

bool pq_mq_busy = false
static

Definition at line 27 of file pqmq.c.

Referenced by mq_putmessage().

◆ pq_mq_handle

shm_mq_handle* pq_mq_handle
static

Definition at line 26 of file pqmq.c.

Referenced by mq_putmessage(), pq_cleanup_redirect_to_shm_mq(), and pq_redirect_to_shm_mq().

◆ pq_mq_parallel_leader_pid

pid_t pq_mq_parallel_leader_pid = 0
static

Definition at line 28 of file pqmq.c.

Referenced by mq_putmessage(), and pq_set_parallel_leader().

◆ pq_mq_parallel_leader_proc_number

pid_t pq_mq_parallel_leader_proc_number = INVALID_PROC_NUMBER
static

Definition at line 29 of file pqmq.c.

Referenced by mq_putmessage(), and pq_set_parallel_leader().

◆ PqCommMqMethods

const PQcommMethods PqCommMqMethods
static
Initial value:
= {
.comm_reset = mq_comm_reset,
.flush = mq_flush,
.flush_if_writable = mq_flush_if_writable,
.is_send_pending = mq_is_send_pending,
.putmessage = mq_putmessage,
.putmessage_noblock = mq_putmessage_noblock
}
static int mq_flush(void)
Definition: pqmq.c:92
static int mq_putmessage(char msgtype, const char *s, size_t len)
Definition: pqmq.c:118
static bool mq_is_send_pending(void)
Definition: pqmq.c:106
static void mq_putmessage_noblock(char msgtype, const char *s, size_t len)
Definition: pqmq.c:199
static void mq_comm_reset(void)
Definition: pqmq.c:86
static int mq_flush_if_writable(void)
Definition: pqmq.c:99

Definition at line 39 of file pqmq.c.

Referenced by pq_redirect_to_shm_mq(), and pq_set_parallel_leader().