mirror/postgresql
2
0
Fork 0
mirror of https://git.postgresql.org/git/postgresql.git synced 2025-02-05 19:09:58 +08:00
Code Issues Packages Projects Releases Wiki Activity

Support frontend-backend protocol communication using a shm_mq.

Browse Source 
A background worker can use pq_redirect_to_shm_mq() to direct protocol
that would normally be sent to the frontend to a shm_mq so that another
process may read them.

The receiving process may use pq_parse_errornotice() to parse an
ErrorResponse or NoticeResponse from the background worker and, if
it wishes, ThrowErrorData() to propagate the error (with or without
further modification).

Patch by me.  Review by Andres Freund.
This commit is contained in:
Robert Haas 2014-10-31 12:02:40 -04:00
parent 252e652ede
commit 2bd9e412f9
9 changed files with 428 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/backend/libpq/Makefile
View File

@ -15,7 +15,7 @@ include $(top_builddir)/src/Makefile.global
# be-fsstubs is here for historical reasons, probably belongs elsewhere
OBJS = be-fsstubs.o be-secure.o auth.o crypt.o hba.o ip.o md5.o pqcomm.o \
pqformat.o pqsignal.o
pqformat.o pqmq.o pqsignal.o
ifeq ($(with_openssl),yes)
OBJS += be-secure-openssl.o

98
src/backend/libpq/pqcomm.c
View File

@ -102,7 +102,6 @@
int Unix_socket_permissions;
char *Unix_socket_group;
/* Where the Unix socket files are (list of palloc'd strings) */
static List *sock_paths = NIL;
@ -134,16 +133,38 @@ static bool DoingCopyOut;
/* Internal functions */
static void pq_close(int code, Datum arg);
static void socket_comm_reset(void);
static void socket_close(int code, Datum arg);
static void socket_set_nonblocking(bool nonblocking);
static int socket_flush(void);
static int socket_flush_if_writable(void);
static bool socket_is_send_pending(void);
static int socket_putmessage(char msgtype, const char *s, size_t len);
static void socket_putmessage_noblock(char msgtype, const char *s, size_t len);
static void socket_startcopyout(void);
static void socket_endcopyout(bool errorAbort);
static int internal_putbytes(const char *s, size_t len);
static int internal_flush(void);
static void pq_set_nonblocking(bool nonblocking);
static void socket_set_nonblocking(bool nonblocking);
#ifdef HAVE_UNIX_SOCKETS
static int Lock_AF_UNIX(char *unixSocketDir, char *unixSocketPath);
static int Setup_AF_UNIX(char *sock_path);
#endif /* HAVE_UNIX_SOCKETS */
PQcommMethods PQcommSocketMethods;
static PQcommMethods PqCommSocketMethods = {
socket_comm_reset,
socket_flush,
socket_flush_if_writable,
socket_is_send_pending,
socket_putmessage,
socket_putmessage_noblock,
socket_startcopyout,
socket_endcopyout
};
/* --------------------------------
* pq_init - initialize libpq at backend startup
@ -152,24 +173,25 @@ static int Setup_AF_UNIX(char *sock_path);
void
pq_init(void)
{
PqCommMethods = &PqCommSocketMethods;
PqSendBufferSize = PQ_SEND_BUFFER_SIZE;
PqSendBuffer = MemoryContextAlloc(TopMemoryContext, PqSendBufferSize);
PqSendPointer = PqSendStart = PqRecvPointer = PqRecvLength = 0;
PqCommBusy = false;
DoingCopyOut = false;
on_proc_exit(pq_close, 0);
on_proc_exit(socket_close, 0);
}
/* --------------------------------
* pq_comm_reset - reset libpq during error recovery
* socket_comm_reset - reset libpq during error recovery
*
* This is called from error recovery at the outer idle loop. It's
* just to get us out of trouble if we somehow manage to elog() from
* inside a pqcomm.c routine (which ideally will never happen, but...)
* --------------------------------
*/
void
pq_comm_reset(void)
static void
socket_comm_reset(void)
{
/* Do not throw away pending data, but do reset the busy flag */
PqCommBusy = false;
@ -178,14 +200,14 @@ pq_comm_reset(void)
}
/* --------------------------------
* pq_close - shutdown libpq at backend exit
* socket_close - shutdown libpq at backend exit
*
* Note: in a standalone backend MyProcPort will be null,
* don't crash during exit...
* --------------------------------
*/
static void
pq_close(int code, Datum arg)
socket_close(int code, Datum arg)
{
if (MyProcPort != NULL)
{
@ -783,15 +805,20 @@ TouchSocketFiles(void)
*/
/* --------------------------------
* pq_set_nonblocking - set socket blocking/non-blocking
* socket_set_nonblocking - set socket blocking/non-blocking
*
* Sets the socket non-blocking if nonblocking is TRUE, or sets it
* blocking otherwise.
* --------------------------------
*/
static void
pq_set_nonblocking(bool nonblocking)
socket_set_nonblocking(bool nonblocking)
{
if (MyProcPort == NULL)
ereport(ERROR,
(errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST),
errmsg("there is no client connection")));
if (MyProcPort->noblock == nonblocking)
return;
@ -844,7 +871,7 @@ pq_recvbuf(void)
}
/* Ensure that we're in blocking mode */
pq_set_nonblocking(false);
socket_set_nonblocking(false);
/* Can fill buffer from PqRecvLength and upwards */
for (;;)
@ -935,7 +962,7 @@ pq_getbyte_if_available(unsigned char *c)
}
/* Put the socket into non-blocking mode */
pq_set_nonblocking(true);
socket_set_nonblocking(true);
r = secure_read(MyProcPort, c, 1);
if (r < 0)
@ -1194,7 +1221,7 @@ internal_putbytes(const char *s, size_t len)
/* If buffer is full, then flush it out */
if (PqSendPointer >= PqSendBufferSize)
{
pq_set_nonblocking(false);
socket_set_nonblocking(false);
if (internal_flush())
return EOF;
}
@ -1210,13 +1237,13 @@ internal_putbytes(const char *s, size_t len)
}
/* --------------------------------
* pq_flush - flush pending output
* socket_flush - flush pending output
*
* returns 0 if OK, EOF if trouble
* --------------------------------
*/
int
pq_flush(void)
static int
socket_flush(void)
{
int res;
@ -1224,7 +1251,7 @@ pq_flush(void)
if (PqCommBusy)
return 0;
PqCommBusy = true;
pq_set_nonblocking(false);
socket_set_nonblocking(false);
res = internal_flush();
PqCommBusy = false;
return res;
@ -1310,8 +1337,8 @@ internal_flush(void)
* Returns 0 if OK, or EOF if trouble.
* --------------------------------
*/
int
pq_flush_if_writable(void)
static int
socket_flush_if_writable(void)
{
int res;
@ -1324,7 +1351,7 @@ pq_flush_if_writable(void)
return 0;
/* Temporarily put the socket into non-blocking mode */
pq_set_nonblocking(true);
socket_set_nonblocking(true);
PqCommBusy = true;
res = internal_flush();
@ -1333,11 +1360,11 @@ pq_flush_if_writable(void)
}
/* --------------------------------
* pq_is_send_pending - is there any pending data in the output buffer?
* socket_is_send_pending - is there any pending data in the output buffer?
* --------------------------------
*/
bool
pq_is_send_pending(void)
static bool
socket_is_send_pending(void)
{
return (PqSendStart < PqSendPointer);
}
@ -1351,7 +1378,7 @@ pq_is_send_pending(void)
/* --------------------------------
* pq_putmessage - send a normal message (suppressed in COPY OUT mode)
* socket_putmessage - send a normal message (suppressed in COPY OUT mode)
*
* If msgtype is not '\0', it is a message type code to place before
* the message body. If msgtype is '\0', then the message has no type
@ -1375,8 +1402,8 @@ pq_is_send_pending(void)
* returns 0 if OK, EOF if trouble
* --------------------------------
*/
int
pq_putmessage(char msgtype, const char *s, size_t len)
static int
socket_putmessage(char msgtype, const char *s, size_t len)
{
if (DoingCopyOut || PqCommBusy)
return 0;
@ -1408,8 +1435,8 @@ fail:
* If the output buffer is too small to hold the message, the buffer
* is enlarged.
*/
void
pq_putmessage_noblock(char msgtype, const char *s, size_t len)
static void
socket_putmessage_noblock(char msgtype, const char *s, size_t len)
{
int res PG_USED_FOR_ASSERTS_ONLY;
int required;
@ -1431,18 +1458,18 @@ pq_putmessage_noblock(char msgtype, const char *s, size_t len)
/* --------------------------------
* pq_startcopyout - inform libpq that an old-style COPY OUT transfer
* socket_startcopyout - inform libpq that an old-style COPY OUT transfer
* is beginning
* --------------------------------
*/
void
pq_startcopyout(void)
static void
socket_startcopyout(void)
{
DoingCopyOut = true;
}
/* --------------------------------
* pq_endcopyout - end an old-style COPY OUT transfer
* socket_endcopyout - end an old-style COPY OUT transfer
*
* If errorAbort is indicated, we are aborting a COPY OUT due to an error,
* and must send a terminator line. Since a partial data line might have
@ -1451,8 +1478,8 @@ pq_startcopyout(void)
* not allow binary transfers, so a textual terminator is always correct.
* --------------------------------
*/
void
pq_endcopyout(bool errorAbort)
static void
socket_endcopyout(bool errorAbort)
{
if (!DoingCopyOut)
return;
@ -1462,7 +1489,6 @@ pq_endcopyout(bool errorAbort)
DoingCopyOut = false;
}
/*
* Support for TCP Keepalive parameters
*/

261
src/backend/libpq/pqmq.c Normal file
View File

@ -0,0 +1,261 @@
/*-------------------------------------------------------------------------
*
* pqmq.c
* Use the frontend/backend protocol for communication over a shm_mq
*
* Portions Copyright (c) 1996-2014, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
* src/backend/libpq/pqmq.c
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "libpq/pqmq.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
static shm_mq *pq_mq;
static shm_mq_handle *pq_mq_handle;
static bool pq_mq_busy = false;
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);
static PQcommMethods PqCommMqMethods = {
mq_comm_reset,
mq_flush,
mq_flush_if_writable,
mq_is_send_pending,
mq_putmessage,
mq_putmessage_noblock,
mq_startcopyout,
mq_endcopyout
};
/*
* Arrange to redirect frontend/backend protocol messages to a shared-memory
* message queue.
*/
void
pq_redirect_to_shm_mq(shm_mq *mq, shm_mq_handle *mqh)
{
PqCommMethods = &PqCommMqMethods;
pq_mq = mq;
pq_mq_handle = mqh;
whereToSendOutput = DestRemote;
FrontendProtocol = PG_PROTOCOL_LATEST;
}
static void
mq_comm_reset(void)
{
/* Nothing to do. */
}
static int
mq_flush(void)
{
/* Nothing to do. */
return 0;
}
static int
mq_flush_if_writable(void)
{
/* Nothing to do. */
return 0;
}
static bool
mq_is_send_pending(void)
{
/* There's never anything pending. */
return 0;
}
/*
* Transmit a libpq protocol message to the shared memory message queue
* selected via pq_mq_handle. We don't include a length word, because the
* receiver will know the length of the message from shm_mq_receive().
*/
static int
mq_putmessage(char msgtype, const char *s, size_t len)
{
shm_mq_iovec iov[2];
shm_mq_result result;
/*
* If we're sending a message, and we have to wait because the
* queue is full, and then we get interrupted, and that interrupt
* results in trying to send another message, we respond by detaching
* the queue. There's no way to return to the original context, but
* even if there were, just queueing the message would amount to
* indefinitely postponing the response to the interrupt. So we do
* this instead.
*/
if (pq_mq_busy)
{
if (pq_mq != NULL)
shm_mq_detach(pq_mq);
pq_mq = NULL;
return EOF;
}
pq_mq_busy = true;
iov[0].data = &msgtype;
iov[0].len = 1;
iov[1].data = s;
iov[1].len = len;
Assert(pq_mq_handle != NULL);
result = shm_mq_sendv(pq_mq_handle, iov, 2, false);
pq_mq_busy = false;
Assert(result == SHM_MQ_SUCCESS || result == SHM_MQ_DETACHED);
if (result != SHM_MQ_SUCCESS)
return EOF;
return 0;
}
static void
mq_putmessage_noblock(char msgtype, const char *s, size_t len)
{
/*
* While the shm_mq machinery does support sending a message in
* non-blocking mode, there's currently no way to try sending beginning
* to send the message that doesn't also commit us to completing the
* transmission. This could be improved in the future, but for now
* we don't need it.
*/
elog(ERROR, "not currently supported");
}
static void
mq_startcopyout(void)
{
/* Nothing to do. */
}
static void
mq_endcopyout(bool errorAbort)
{
/* Nothing to do. */
}
/*
* Parse an ErrorResponse or NoticeResponse payload and populate an ErrorData
* structure with the results.
*/
void
pq_parse_errornotice(StringInfo msg, ErrorData *edata)
{
/* Initialize edata with reasonable defaults. */
MemSet(edata, 0, sizeof(ErrorData));
edata->elevel = ERROR;
edata->assoc_context = CurrentMemoryContext;
/* Loop over fields and extract each one. */
for (;;)
{
char code = pq_getmsgbyte(msg);
const char *value;
if (code == '\0')
{
pq_getmsgend(msg);
break;
}
value = pq_getmsgstring(msg);
switch (code)
{
case PG_DIAG_SEVERITY:
if (strcmp(value, "DEBUG") == 0)
edata->elevel = DEBUG1; /* or some other DEBUG level */
else if (strcmp(value, "LOG") == 0)
edata->elevel = LOG; /* can't be COMMERROR */
else if (strcmp(value, "INFO") == 0)
edata->elevel = INFO;
else if (strcmp(value, "NOTICE") == 0)
edata->elevel = NOTICE;
else if (strcmp(value, "WARNING") == 0)
edata->elevel = WARNING;
else if (strcmp(value, "ERROR") == 0)
edata->elevel = ERROR;
else if (strcmp(value, "FATAL") == 0)
edata->elevel = FATAL;
else if (strcmp(value, "PANIC") == 0)
edata->elevel = PANIC;
else
elog(ERROR, "unknown error severity");
break;
case PG_DIAG_SQLSTATE:
if (strlen(value) != 5)
elog(ERROR, "malformed sql state");
edata->sqlerrcode = MAKE_SQLSTATE(value[0], value[1], value[2],
value[3], value[4]);
break;
case PG_DIAG_MESSAGE_PRIMARY:
edata->message = pstrdup(value);
break;
case PG_DIAG_MESSAGE_DETAIL:
edata->detail = pstrdup(value);
break;
case PG_DIAG_MESSAGE_HINT:
edata->hint = pstrdup(value);
break;
case PG_DIAG_STATEMENT_POSITION:
edata->cursorpos = pg_atoi(value, sizeof(int), '\0');
break;
case PG_DIAG_INTERNAL_POSITION:
edata->internalpos = pg_atoi(value, sizeof(int), '\0');
break;
case PG_DIAG_INTERNAL_QUERY:
edata->internalquery = pstrdup(value);
break;
case PG_DIAG_CONTEXT:
edata->context = pstrdup(value);
break;
case PG_DIAG_SCHEMA_NAME:
edata->schema_name = pstrdup(value);
break;
case PG_DIAG_TABLE_NAME:
edata->table_name = pstrdup(value);
break;
case PG_DIAG_COLUMN_NAME:
edata->column_name = pstrdup(value);
break;
case PG_DIAG_DATATYPE_NAME:
edata->datatype_name = pstrdup(value);
break;
case PG_DIAG_CONSTRAINT_NAME:
edata->constraint_name = pstrdup(value);
break;
case PG_DIAG_SOURCE_FILE:
edata->filename = pstrdup(value);
break;
case PG_DIAG_SOURCE_LINE:
edata->lineno = pg_atoi(value, sizeof(int), '\0');
break;
case PG_DIAG_SOURCE_FUNCTION:
edata->funcname = pstrdup(value);
break;
default:
elog(ERROR, "unknown error field: %d", (int) code);
break;
}
}
}

2
src/backend/utils/adt/numutils.c
View File

@ -34,7 +34,7 @@
* overflow.
*/
int32
pg_atoi(char *s, int size, int c)
pg_atoi(const char *s, int size, int c)
{
long l;
char *badp;

51
src/backend/utils/error/elog.c
View File

@ -1576,6 +1576,57 @@ FlushErrorState(void)
MemoryContextResetAndDeleteChildren(ErrorContext);
}
/*
* ThrowErrorData --- report an error described by an ErrorData structure
*
* This is intended to be used to re-report errors originally thrown by
* background worker processes and then propagated (with or without
* modification) to the backend responsible for them.
*/
void
ThrowErrorData(ErrorData *edata)
{
ErrorData *newedata;
MemoryContext oldcontext;
if (!errstart(edata->elevel, edata->filename, edata->lineno,
edata->funcname, NULL))
return;
newedata = &errordata[errordata_stack_depth];
oldcontext = MemoryContextSwitchTo(edata->assoc_context);
/* Copy the supplied fields to the error stack. */
if (edata->sqlerrcode > 0)
newedata->sqlerrcode = edata->sqlerrcode;
if (edata->message)
newedata->message = pstrdup(edata->message);
if (edata->detail)
newedata->detail = pstrdup(edata->detail);
if (edata->detail_log)
newedata->detail_log = pstrdup(edata->detail_log);
if (edata->hint)
newedata->hint = pstrdup(edata->hint);
if (edata->context)
newedata->context = pstrdup(edata->context);
if (edata->schema_name)
newedata->schema_name = pstrdup(edata->schema_name);
if (edata->table_name)
newedata->table_name = pstrdup(edata->table_name);
if (edata->column_name)
newedata->column_name = pstrdup(edata->column_name);
if (edata->datatype_name)
newedata->datatype_name = pstrdup(edata->datatype_name);
if (edata->constraint_name)
newedata->constraint_name = pstrdup(edata->constraint_name);
if (edata->internalquery)
newedata->internalquery = pstrdup(edata->internalquery);
MemoryContextSwitchTo(oldcontext);
errfinish(0);
}
/*
* ReThrowError --- re-throw a previously copied error
*

36
src/include/libpq/libpq.h
View File

@ -37,6 +37,31 @@ typedef struct
} u;
} PQArgBlock;
typedef struct
{
void (*comm_reset)(void);
int (*flush)(void);
int (*flush_if_writable)(void);
bool (*is_send_pending)(void);
int (*putmessage)(char msgtype, const char *s, size_t len);
void (*putmessage_noblock)(char msgtype, const char *s, size_t len);
void (*startcopyout)(void);
void (*endcopyout)(bool errorAbort);
} PQcommMethods;
PQcommMethods *PqCommMethods;
#define pq_comm_reset() (PqCommMethods->comm_reset())
#define pq_flush() (PqCommMethods->flush())
#define pq_flush_if_writable() (PqCommMethods->flush_if_writable())
#define pq_is_send_pending() (PqCommMethods->is_send_pending())
#define pq_putmessage(msgtype, s, len) \
(PqCommMethods->putmessage(msgtype, s, len))
#define pq_putmessage_noblock(msgtype, s, len) \
(PqCommMethods->putmessage(msgtype, s, len))
#define pq_startcopyout() (PqCommMethods->startcopyout())
#define pq_endcopyout(errorAbort) (PqCommMethods->endcopyout(errorAbort))
/*
* External functions.
*/
@ -51,7 +76,6 @@ extern int StreamConnection(pgsocket server_fd, Port *port);
extern void StreamClose(pgsocket sock);
extern void TouchSocketFiles(void);
extern void pq_init(void);
extern void pq_comm_reset(void);
extern int pq_getbytes(char *s, size_t len);
extern int pq_getstring(StringInfo s);
extern int pq_getmessage(StringInfo s, int maxlen);
@ -59,13 +83,6 @@ extern int pq_getbyte(void);
extern int pq_peekbyte(void);
extern int pq_getbyte_if_available(unsigned char *c);
extern int pq_putbytes(const char *s, size_t len);
extern int pq_flush(void);
extern int pq_flush_if_writable(void);
extern bool pq_is_send_pending(void);
extern int pq_putmessage(char msgtype, const char *s, size_t len);
extern void pq_putmessage_noblock(char msgtype, const char *s, size_t len);
extern void pq_startcopyout(void);
extern void pq_endcopyout(bool errorAbort);
/*
* prototypes for functions in be-secure.c
@ -75,6 +92,9 @@ extern char *ssl_key_file;
extern char *ssl_ca_file;
extern char *ssl_crl_file;
extern int (*pq_putmessage_hook)(char msgtype, const char *s, size_t len);
extern int (*pq_flush_hook)(void);
extern int secure_initialize(void);
extern bool secure_loaded_verify_locations(void);
extern void secure_destroy(void);

22
src/include/libpq/pqmq.h Normal file
View File

@ -0,0 +1,22 @@
/*-------------------------------------------------------------------------
*
* pqmq.h
* Use the frontend/backend protocol for communication over a shm_mq
*
* Portions Copyright (c) 1996-2014, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
* src/include/libpq/pqmq.h
*
*-------------------------------------------------------------------------
*/
#ifndef PQMQ_H
#define PQMQ_H
#include "storage/shm_mq.h"
extern void pq_redirect_to_shm_mq(shm_mq *, shm_mq_handle *);
extern void pq_parse_errornotice(StringInfo str, ErrorData *edata);
#endif /* PQMQ_H */

2
src/include/utils/builtins.h
View File

@ -285,7 +285,7 @@ extern Datum current_schema(PG_FUNCTION_ARGS);
extern Datum current_schemas(PG_FUNCTION_ARGS);
/* numutils.c */
extern int32 pg_atoi(char *s, int size, int c);
extern int32 pg_atoi(const char *s, int size, int c);
extern void pg_itoa(int16 i, char *a);
extern void pg_ltoa(int32 l, char *a);
extern void pg_lltoa(int64 ll, char *a);

1
src/include/utils/elog.h
View File

@ -415,6 +415,7 @@ extern ErrorData *CopyErrorData(void);
extern void FreeErrorData(ErrorData *edata);
extern void FlushErrorState(void);
extern void ReThrowError(ErrorData *edata) __attribute__((noreturn));
extern void ThrowErrorData(ErrorData *edata);
extern void pg_re_throw(void) __attribute__((noreturn));
extern char *GetErrorContextStack(void);