pgpool2.git / summary / log / commit / refs
commit e14a25a33fcdd0df70a3abcbdb89cb90cfa6b718
Author: Muhammad Usama <m.usama@gmail.com>
Date: Sun Nov 24 18:11:40 2013 +0000
Some more on integration of exception and memory manager front.
The current checkin includes creation of per loop iteration memory context for
worker child process, and integration of above in some parts of query processing logics.
For the query processing function all my current code browsing leads me to the
believe that although "pool_process_query()" caller has handled five possible
return values from the function i.e POOL_END, POOL_ERROR, POOL_FATAL,
POOL_IDLE and POOL_CONTINUE, but in actual the function returns only POOL_END,
POOL_ERROR and the success which is POOL_CONTINUE. and to squeeze in the managers
in query processing function I am moving forward with the above mentioned assumption.
Finally the checkin puts the wrapper on pool_read function, "pool_read_with_error()"
which throws an ereport instead of returning -ve status in case of an error.
And this wrapper will live till the time when all the callers of pool_read()
will be decorated with the exception manager.
src/context/pool_session_context.c | 2 +-
src/include/pool.h | 3 +-
src/include/utils/pool_stream.h | 3 +
src/protocol/child.c | 249 ++++++++++----------------
src/protocol/pool_process_query.c | 62 ++-----
src/streaming_replication/pool_worker_child.c | 107 ++++++++---
src/utils/error/elog.c | 11 ++
src/utils/pool_stream.c | 11 ++
8 files changed, 221 insertions(+), 227 deletions(-)
diff --git a/src/context/pool_session_context.c b/src/context/pool_session_context.c
index 7b785be2d..f1b8f0ac7 100644
--- a/src/context/pool_session_context.c
+++ b/src/context/pool_session_context.c
@@ -64,7 +64,7 @@ void pool_init_session_context(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *
init_sent_message_list();
/* Create memory context */
- /* TODO rething about the parent for this context ??*/
+ /* TODO re-think about the parent for this context ??*/
session_context->memory_context = AllocSetContextCreate(TopMemoryContext,
"SessionContext",
ALLOCSET_SMALL_MINSIZE,
diff --git a/src/include/pool.h b/src/include/pool.h
index c7da1fe68..d0135552d 100644
--- a/src/include/pool.h
+++ b/src/include/pool.h
@@ -589,7 +589,7 @@ extern void ClientAuthentication(POOL_CONNECTION *frontend);
extern void pool_getnameinfo_all(SockAddr *saddr, char *remote_host, char *remote_port);
/* strlcpy.c */
-extern size_t strlcpy(char *dst, const char *src, size_t siz);
+//extern size_t strlcpy(char *dst, const char *src, size_t siz);
/* ps_status.c */
extern bool update_process_title;
@@ -609,6 +609,7 @@ extern int wait_connection_closed(void);
extern void cancel_request(CancelPacket *sp);
extern void check_stop_request(void);
extern void pool_initialize_private_backend_status(void);
+extern bool is_session_connected(void);
/* pool_process_query.c */
extern void reset_variables(void);
diff --git a/src/include/utils/pool_stream.h b/src/include/utils/pool_stream.h
index d7e15ea8a..2cabe4e8e 100644
--- a/src/include/utils/pool_stream.h
+++ b/src/include/utils/pool_stream.h
@@ -45,6 +45,9 @@
extern POOL_CONNECTION *pool_open(int fd);
extern void pool_close(POOL_CONNECTION *cp);
extern int pool_read(POOL_CONNECTION *cp, void *buf, int len);
+extern void pool_read_with_error(POOL_CONNECTION *cp, void *buf, int len,
+ const char* err_context );
+
extern char *pool_read2(POOL_CONNECTION *cp, int len);
extern int pool_write(POOL_CONNECTION *cp, void *buf, int len);
extern int pool_flush(POOL_CONNECTION *cp);
diff --git a/src/protocol/child.c b/src/protocol/child.c
index 778bd2365..3ef1c5bfe 100644
--- a/src/protocol/child.c
+++ b/src/protocol/child.c
@@ -771,9 +771,8 @@ static StartupPacket *read_startup_packet(POOL_CONNECTION *cp)
enable_authentication_timeout();
/* read startup packet length */
- if (pool_read(cp, &len, sizeof(len)))
- ereport(ERROR,
- (errmsg("Invalid startup packet")));
+ pool_read_with_error(cp, &len, sizeof(len),
+ "startup packet length");
len = ntohl(len);
len -= sizeof(len);
@@ -785,9 +784,8 @@ static StartupPacket *read_startup_packet(POOL_CONNECTION *cp)
sp->startup_packet = palloc0(len);
/* read startup packet */
- if (pool_read(cp, sp->startup_packet, len))
- ereport(ERROR,
- (errmsg("Invalid startup packet")));
+ pool_read_with_error(cp, sp->startup_packet, len,
+ "startup packet");
sp->len = len;
memcpy(&protov, sp->startup_packet, sizeof(protov));
@@ -838,7 +836,8 @@ static StartupPacket *read_startup_packet(POOL_CONNECTION *cp)
{
p += (strlen(p) + 1);
sp->application_name = p;
- pool_debug("read_startup_packet: application_name: %s", p);
+ ereport(DEBUG1,
+ (errmsg("read_startup_packet: application_name: %s", p)));
}
p += (strlen(p) + 1);
@@ -997,7 +996,8 @@ void cancel_request(CancelPacket *sp)
CancelPacket cp;
bool found = false;
- pool_debug("Cancel request received");
+ ereport(DEBUG1,
+ (errmsg("Cancel request received")));
/* look for cancel key from shmem info */
for (i=0;i<pool_config->num_init_children;i++)
@@ -1397,21 +1397,8 @@ POOL_CONNECTION_POOL_SLOT *make_persistent_db_connection(
int len, len1;
int status;
- cp = malloc(sizeof(POOL_CONNECTION_POOL_SLOT));
- if (cp == NULL)
- {
- pool_error("make_persistent_db_connection: could not allocate memory");
- return NULL;
- }
- memset(cp, 0, sizeof(POOL_CONNECTION_POOL_SLOT));
-
- startup_packet = malloc(sizeof(*startup_packet));
- if (startup_packet == NULL)
- {
- pool_error("make_persistent_db_connection: could not allocate memory");
- return NULL;
- }
- memset(startup_packet, 0, sizeof(*startup_packet));
+ cp = palloc0(sizeof(POOL_CONNECTION_POOL_SLOT));
+ startup_packet = palloc0(sizeof(*startup_packet));
startup_packet->protoVersion = htonl(0x00030000); /* set V3 proto major/minor */
/*
@@ -1428,8 +1415,10 @@ POOL_CONNECTION_POOL_SLOT *make_persistent_db_connection(
if (fd < 0)
{
- pool_error("make_persistent_db_connection: connection to %s(%d) failed", hostname, port);
- return NULL;
+ ereport(ERROR,
+ (errmsg("failed to make persistent db connection"),
+ errdetail("connection to %s(%d) failed", hostname, port)));
+
}
cp->con = pool_open(fd);
@@ -1444,51 +1433,39 @@ POOL_CONNECTION_POOL_SLOT *make_persistent_db_connection(
len1 = snprintf(&startup_packet->data[len], sizeof(startup_packet->data)-len, "%s", user) + 1;
if (len1 >= (sizeof(startup_packet->data)-len))
{
- pool_error("make_persistent_db_connection: too long user name");
- return NULL;
+ ereport(ERROR,
+ (errmsg("failed to make persistent db connection"),
+ errdetail("user name is too long")));
}
len += len1;
len1 = snprintf(&startup_packet->data[len], sizeof(startup_packet->data)-len, "database") + 1;
if (len1 >= (sizeof(startup_packet->data)-len))
{
- pool_error("make_persistent_db_connection: too long user name");
- return NULL;
+ ereport(ERROR,
+ (errmsg("failed to make persistent db connection"),
+ errdetail("user name is too long")));
}
len += len1;
len1 = snprintf(&startup_packet->data[len], sizeof(startup_packet->data)-len, "%s", dbname) + 1;
if (len1 >= (sizeof(startup_packet->data)-len))
{
- pool_error("make_persistent_db_connection: too long database name");
- return NULL;
+ ereport(ERROR,
+ (errmsg("failed to make persistent db connection"),
+ errdetail("database name is too long")));
}
len += len1;
startup_packet->data[len++] = '\0';
- cp->sp = malloc(sizeof(StartupPacket));
- if (cp->sp == NULL)
- {
- pool_error("make_persistent_db_connection: could not allocate memory");
- return NULL;
- }
+ cp->sp = palloc(sizeof(StartupPacket));
cp->sp->startup_packet = (char *)startup_packet;
cp->sp->len = len + 4;
cp->sp->major = 3;
cp->sp->minor = 0;
- cp->sp->database = strdup(dbname);
- if (cp->sp->database == NULL)
- {
- pool_error("make_persistent_db_connection: could not allocate memory");
- return NULL;
- }
- cp->sp->user = strdup(user);
- if (cp->sp->user == NULL)
- {
- pool_error("make_persistent_db_connection: could not allocate memory");
- return NULL;
- }
+ cp->sp->database = pstrdup(dbname);
+ cp->sp->user = pstrdup(user);
/*
* send startup packet
@@ -1496,8 +1473,10 @@ POOL_CONNECTION_POOL_SLOT *make_persistent_db_connection(
status = send_startup_packet(cp);
if (status)
{
- pool_error("make_persistent_db_connection: send_startup_packet failed");
- return NULL;
+ ereport(ERROR,
+ (errmsg("failed to make persistent db connection"),
+ errdetail("unable to send startup packet")));
+
}
/*
@@ -1505,8 +1484,9 @@ POOL_CONNECTION_POOL_SLOT *make_persistent_db_connection(
*/
if (s_do_auth(cp, password))
{
- pool_error("make_persistent_db_connection: s_do_auth failed");
- return NULL;
+ ereport(ERROR,
+ (errmsg("failed to make persistent db connection"),
+ errdetail("authentication failed")));
}
return cp;
@@ -1539,11 +1519,11 @@ void discard_persistent_db_connection(POOL_CONNECTION_POOL_SLOT *cp)
pool_unset_nonblock(cp->con->fd);
pool_close(cp->con);
- free(cp->sp->startup_packet);
- free(cp->sp->database);
- free(cp->sp->user);
- free(cp->sp);
- free(cp);
+ pfree(cp->sp->startup_packet);
+ pfree(cp->sp->database);
+ pfree(cp->sp->user);
+ pfree(cp->sp);
+ pfree(cp);
}
/*
@@ -1564,37 +1544,28 @@ static int s_do_auth(POOL_CONNECTION_POOL_SLOT *cp, char *password)
/*
* read kind expecting 'R' packet (authentication response)
*/
- status = pool_read(cp->con, &kind, sizeof(kind));
- if (status < 0)
- {
- pool_error("s_do_auth: error while reading message kind");
- return -1;
- }
+ pool_read_with_error(cp->con, &kind, sizeof(kind),
+ "authentication message response type");
if (kind != 'R')
{
- pool_error("s_do_auth: expecting R got %c", kind);
- return -1;
+ ereport(ERROR,
+ (errmsg("failed to authenticate"),
+ errdetail("invalid authentication message response type, Expecting 'R' and received '%c'",kind)));
}
/* read message length */
- status = pool_read(cp->con, &length, sizeof(length));
- if (status < 0)
- {
- pool_error("s_do_auth: error while reading message length");
- return -1;
- }
+ pool_read_with_error(cp->con, &length, sizeof(length),
+ "authentication message response length");
+
length = ntohl(length);
/* read auth kind */
- status = pool_read(cp->con, &auth_kind, sizeof(auth_kind));
- if (status < 0)
- {
- pool_error("s_do_auth: error while reading auth kind");
- return -1;
- }
+ pool_read_with_error(cp->con, &auth_kind, sizeof(auth_kind),
+ "authentication method from response");
auth_kind = ntohl(auth_kind);
- pool_debug("s_do_auth: auth kind: %d", auth_kind);
+ ereport(DEBUG1,
+ (errmsg("authenticate kind = %d",auth_kind)));
if (auth_kind == 0) /* trust authentication? */
{
@@ -1610,8 +1581,9 @@ static int s_do_auth(POOL_CONNECTION_POOL_SLOT *cp, char *password)
status = pool_flush(cp->con);
if (status > 0)
{
- pool_error("s_do_auth: error while sending clear text password");
- return -1;
+ ereport(ERROR,
+ (errmsg("failed to authenticate"),
+ errdetail("error while sending clear text password")));
}
return s_do_auth(cp, password);
}
@@ -1621,12 +1593,7 @@ static int s_do_auth(POOL_CONNECTION_POOL_SLOT *cp, char *password)
char salt[3];
char *crypt_password;
- status = pool_read(cp->con, &salt, 2);
- if (status > 0)
- {
- pool_error("s_do_auth: error while reading crypt salt");
- return -1;
- }
+ pool_read_with_error(cp->con, &salt, 2,"crypt salt");
salt[2] = '\0';
crypt_password = crypt(password, salt);
@@ -1637,8 +1604,9 @@ static int s_do_auth(POOL_CONNECTION_POOL_SLOT *cp, char *password)
status = pool_flush(cp->con);
if (status > 0)
{
- pool_error("s_do_auth: error while sending crypt password");
- return -1;
+ ereport(ERROR,
+ (errmsg("failed to authenticate"),
+ errdetail("error while sending crypt password")));
}
return s_do_auth(cp, password);
}
@@ -1648,20 +1616,9 @@ static int s_do_auth(POOL_CONNECTION_POOL_SLOT *cp, char *password)
char *buf, *buf1;
int size;
- status = pool_read(cp->con, &salt, 4);
- if (status > 0)
- {
- pool_error("s_do_auth: error while reading md5 salt");
- return -1;
- }
+ pool_read_with_error(cp->con, &salt, 4,"authentication md5 salt");
- buf = malloc(2 * (MD5_PASSWD_LEN + 4)); /* hash + "md5" + '\0' */
- if (buf == NULL)
- {
- pool_error("s_do_auth(): malloc failed: %s", strerror(errno));
- return -1;
- }
- memset(buf, 0, 2 * (MD5_PASSWD_LEN + 4));
+ buf = palloc0(2 * (MD5_PASSWD_LEN + 4)); /* hash + "md5" + '\0' */
/* build md5 password */
buf1 = buf + MD5_PASSWD_LEN + 4;
@@ -1676,18 +1633,20 @@ static int s_do_auth(POOL_CONNECTION_POOL_SLOT *cp, char *password)
status = pool_flush(cp->con);
if (status > 0)
{
- pool_error("s_do_auth: error while sending md5 password");
- return -1;
+ ereport(ERROR,
+ (errmsg("failed to authenticate"),
+ errdetail("error while sending md5 password")));
}
status = s_do_auth(cp, password);
- free(buf);
+ pfree(buf);
return status;
}
else
{
- pool_error("s_do_auth: auth kind %d not supported yet", auth_kind);
- return -1;
+ ereport(ERROR,
+ (errmsg("failed to authenticate"),
+ errdetail("auth kind %d not supported yet", auth_kind)));
}
/*
@@ -1699,12 +1658,8 @@ static int s_do_auth(POOL_CONNECTION_POOL_SLOT *cp, char *password)
for (;;)
{
- status = pool_read(cp->con, &kind, sizeof(kind));
- if (status < 0)
- {
- pool_error("s_do_auth: error while reading message kind");
- return -1;
- }
+ pool_read_with_error(cp->con, &kind, sizeof(kind),
+ "authentication message kind");
switch (kind)
{
@@ -1713,52 +1668,32 @@ static int s_do_auth(POOL_CONNECTION_POOL_SLOT *cp, char *password)
pool_debug("s_do_auth: backend key data received");
/* read message length */
- status = pool_read(cp->con, &length, sizeof(length));
- if (status < 0)
- {
- pool_error("s_do_auth: error while reading message length");
- return -1;
- }
+ pool_read_with_error(cp->con, &length, sizeof(length),"message length for authentication kind 'K'");
if (ntohl(length) != 12)
{
pool_error("s_do_auth: backend key data length is not 12 (%d)", ntohl(length));
}
/* read pid */
- if (pool_read(cp->con, &pid, sizeof(pid)) < 0)
- {
- pool_error("s_do_auth: failed to read pid");
- return -1;
- }
+ pool_read_with_error(cp->con, &pid, sizeof(pid),"pid for authentication kind 'K'");
cp->pid = pid;
/* read key */
- if (pool_read(cp->con, &key, sizeof(key)) < 0)
- {
- pool_error("s_do_auth: failed to read key");
- return -1;
- }
+ pool_read_with_error(cp->con, &key, sizeof(key),
+ "key for authentication kind 'K'");
cp->key = key;
break;
case 'Z': /* Ready for query */
/* read message length */
- status = pool_read(cp->con, &length, sizeof(length));
- if (status < 0)
- {
- pool_error("s_do_auth: error while reading message length");
- return -1;
- }
+ pool_read_with_error(cp->con, &length, sizeof(length),
+ "message length for authentication kind 'Z'");
length = ntohl(length);
/* read transaction state */
- status = pool_read(cp->con, &state, sizeof(state));
- if (status < 0)
- {
- pool_error("s_do_auth: error while reading transaction state");
- return -1;
- }
-
+ pool_read_with_error(cp->con, &state, sizeof(state),
+ "transaction state for authentication kind 'Z'");
+
pool_debug("s_do_auth: transaction state: %c", state);
cp->con->tstate = state;
@@ -1773,13 +1708,9 @@ static int s_do_auth(POOL_CONNECTION_POOL_SLOT *cp, char *password)
case 'N': /* notice response */
case 'E': /* error response */
/* Just throw away data */
- status = pool_read(cp->con, &length, sizeof(length));
- if (status < 0)
- {
- pool_error("s_do_auth: error while reading message length. kind:%c", kind);
- return -1;
- }
-
+ pool_read_with_error(cp->con, &length, sizeof(length),
+ "backend message length");
+
length = ntohl(length);
length -= 4;
@@ -1932,7 +1863,10 @@ static void init_system_db_connection(void)
system_db_connect();
if (PQstatus(system_db_info->pgconn) != CONNECTION_OK)
{
- pool_error("Could not make persistent libpq system DB connection");
+ ereport(ERROR,
+ (errmsg("failed to make persistent system db connection"),
+ errdetail("system_db_connect failed")));
+
}
system_db_info->connection = make_persistent_db_connection(pool_config->system_db_hostname,
@@ -1940,10 +1874,6 @@ static void init_system_db_connection(void)
pool_config->system_db_dbname,
pool_config->system_db_user,
pool_config->system_db_password, false);
- if (system_db_info->connection == NULL)
- {
- pool_error("Could not make persistent system DB connection");
- }
}
}
@@ -2399,7 +2329,7 @@ retry_startup:
*/
/* Check if restart request is set because of failback event
- * happend. If so, close idle connections to backend and make
+ * happened. If so, close idle connections to backend and make
* a new copy of backend status.
*/
if (pool_get_my_process_info()->need_to_restart)
@@ -2472,3 +2402,12 @@ static void print_process_status(char *remote_host,char* remote_port)
else
snprintf(remote_ps_data, sizeof(remote_ps_data),"%s(%s)",remote_host, remote_port);
}
+
+bool is_session_connected()
+{
+ if(processType == PT_CHILD)
+ return (pool_get_session_context() != NULL);
+ return false;
+}
+
+
diff --git a/src/protocol/pool_process_query.c b/src/protocol/pool_process_query.c
index b6cfa0923..e332b41dd 100644
--- a/src/protocol/pool_process_query.c
+++ b/src/protocol/pool_process_query.c
@@ -237,7 +237,7 @@ POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
/*
* ProcessFrontendResponse() may start query
- * processing. We need to recheck
+ * processing. We need to re-check
* pool_is_query_in_progress() here.
*/
if (pool_is_query_in_progress())
@@ -300,12 +300,8 @@ POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
/* If master does not have pending
* data, we discard one packet from
* other backend */
- status = pool_read(CONNECTION(backend, i), &kind, sizeof(kind));
- if (status < 0)
- {
- pool_error("pool_process_query: error while reading message kind from backend %d", i);
- return POOL_ERROR;
- }
+ pool_read_with_error(CONNECTION(backend, i), &kind, sizeof(kind),
+ "reading message kind from backend");
if (kind == 'A')
{
@@ -318,11 +314,9 @@ POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
*/
pool_unread(CONNECTION(backend, i), &kind, sizeof(kind));
pool_log("pool_process_query: received %c packet from backend %d. Don't dicard and read %c packet from master", kind, i, kind);
- if (pool_read(CONNECTION(backend, MASTER_NODE_ID), &kind, sizeof(kind)) < 0)
- {
- pool_error("pool_process_query: error while reading message kind from backend %d", MASTER_NODE_ID);
- return POOL_ERROR;
- }
+ pool_read_with_error(CONNECTION(backend, MASTER_NODE_ID), &kind, sizeof(kind),
+ "reading message kind from backend");
+
pool_unread(CONNECTION(backend, MASTER_NODE_ID), &kind, sizeof(kind));
}
else
@@ -331,11 +325,8 @@ POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
if (MAJOR(backend) == PROTO_MAJOR_V3)
{
- if (pool_read(CONNECTION(backend, i), &len, sizeof(len)) < 0)
- {
- pool_error("pool_process_query: error while reading message length from backend %d", i);
- return POOL_ERROR;
- }
+ pool_read_with_error(CONNECTION(backend, i), &len, sizeof(len),
+ "reading message length from backend");
len = ntohl(len) - 4;
string = pool_read2(CONNECTION(backend, i), len);
if (string == NULL)
@@ -973,7 +964,7 @@ int pool_check_fd(POOL_CONNECTION *cp)
/*
* send "terminate"(X) message to all backends, indicating that
* backend should prepare to close connection to frontend (actually
- * pgpool). Note that caller must be protecedt from a signal
+ * pgpool). Note that caller must be proteceted from a signal
* interruption while calling this function. Otherwise the number of
* valid backends might be changed by failover/failback.
*/
@@ -2402,41 +2393,16 @@ POOL_STATUS do_query(POOL_CONNECTION *backend, char *query, POOL_SELECT_RESULT *
pool_debug("do_query: extended:%d query:%s", doing_extended, query);
*result = NULL;
- res = malloc(sizeof(*res));
- if (!res)
- {
- pool_error("pool_query: malloc failed");
- return POOL_ERROR;
- }
- rowdesc = malloc(sizeof(*rowdesc));
- if (!rowdesc)
- {
- if (res)
- free(res);
- pool_error("pool_query: malloc failed");
- return POOL_ERROR;
- }
- memset(res, 0, sizeof(*res));
- memset(rowdesc, 0, sizeof(*rowdesc));
+ res = palloc0(sizeof(*res));
+ rowdesc = palloc0(sizeof(*rowdesc));
*result = res;
res->rowdesc = rowdesc;
num_data = 0;
- res->nullflags = malloc(DO_QUERY_ALLOC_NUM*sizeof(int));
- if (!res->nullflags)
- {
- pool_error("do_query: malloc failed");
- return POOL_ERROR;
- }
- res->data = malloc(DO_QUERY_ALLOC_NUM*sizeof(char *));
- if (!res->data)
- {
- pool_error("do_query: malloc failed");
- return POOL_ERROR;
- }
- memset(res->data, 0, DO_QUERY_ALLOC_NUM*sizeof(char *));
+ res->nullflags = palloc(DO_QUERY_ALLOC_NUM*sizeof(int));
+ res->data = palloc0(DO_QUERY_ALLOC_NUM*sizeof(char *));
/*
* Send a query to the backend. We use extended query proctocol
@@ -4915,7 +4881,7 @@ SELECT_RETRY:
pool_error("connection on node %d was terminated due to conflict with recovery", i);
pool_send_fatal_message(frontend, MAJOR(backend),
SERIALIZATION_FAIL_ERROR_CODE,
- "connection was terminated due to confilict with recovery",
+ "connection was terminated due to conflict with recovery",
"User was holding a relation lock for too long.",
"In a moment you should be able to reconnect to the database and repeat your command.",
__FILE__, __LINE__);
diff --git a/src/streaming_replication/pool_worker_child.c b/src/streaming_replication/pool_worker_child.c
index 093c8cc2e..e17255ac7 100644
--- a/src/streaming_replication/pool_worker_child.c
+++ b/src/streaming_replication/pool_worker_child.c
@@ -23,6 +23,7 @@
*/
#include "config.h"
+
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
@@ -50,6 +51,10 @@
#endif
#include "pool.h"
+#include "utils/palloc.h"
+#include "utils/memutils.h"
+#include "utils/elog.h"
+
#include "context/pool_process_context.h"
#include "context/pool_session_context.h"
#include "pool_config.h"
@@ -72,7 +77,7 @@ static long text_to_lsn(char *text);
static RETSIGTYPE my_signal_handler(int sig);
static RETSIGTYPE reload_config_handler(int sig);
static void reload_config(void);
-
+static void clean_persistent_connection_error_callback(void *arg);
#define CHECK_REQUEST \
do { \
if (reload_config_request) \
@@ -91,6 +96,9 @@ static void reload_config(void);
*/
void do_worker_child(void)
{
+ sigjmp_buf local_sigjmp_buf;
+ MemoryContext WorkerMemoryContext;
+
pool_debug("I am %d", getpid());
/* Identify myself via ps */
@@ -108,14 +116,38 @@ void do_worker_child(void)
signal(SIGUSR2, SIG_IGN);
signal(SIGPIPE, SIG_IGN);
+ /* Create per loop iteration memory context */
+ WorkerMemoryContext = AllocSetContextCreate(TopMemoryContext,
+ "Worker_main_loop",
+ ALLOCSET_DEFAULT_MINSIZE,
+ ALLOCSET_DEFAULT_INITSIZE,
+ ALLOCSET_DEFAULT_MAXSIZE);
+
+ MemoryContextSwitchTo(TopMemoryContext);
+
/* Initialize my backend status */
pool_initialize_private_backend_status();
/* Initialize per process context */
pool_init_process_context();
+ if (sigsetjmp(local_sigjmp_buf, 1) != 0)
+ {
+ pool_signal(SIGALRM, SIG_IGN);
+ error_context_stack = NULL;
+ EmitErrorReport();
+ MemoryContextSwitchTo(TopMemoryContext);
+ FlushErrorState();
+ }
+ /* We can now handle ereport(ERROR) */
+ PG_exception_stack = &local_sigjmp_buf;
+
+
for (;;)
{
+ MemoryContextSwitchTo(WorkerMemoryContext);
+ MemoryContextResetAndDeleteChildren(WorkerMemoryContext);
+
CHECK_REQUEST;
if (pool_config->sr_check_period <= 0)
@@ -129,11 +161,20 @@ void do_worker_child(void)
if (pool_config->sr_check_period > 0 && MASTER_SLAVE && !strcmp(pool_config->master_slave_sub_mode, MODE_STREAMREP))
{
+ ErrorContextCallback errcallback;
/* Check and establish persistent connections to the backend */
establish_persistent_connection();
+ errcallback.callback = clean_persistent_connection_error_callback;
+ errcallback.arg = NULL;
+ errcallback.previous = error_context_stack;
+ error_context_stack = &errcallback;
+
/* Do replication time lag checking */
check_replication_time_lag();
+
+ /* Pop the error context stack */
+ error_context_stack = errcallback.previous;
/* Discard persistent connections */
discard_persistent_connection();
@@ -143,6 +184,15 @@ void do_worker_child(void)
exit(0);
}
+/*
+ * Error context callback for errors occurring after persistent db connection
+ * was created.
+ */
+static void
+clean_persistent_connection_error_callback(void *arg)
+{
+ discard_persistent_connection();
+}
/*
* Establish persistent connection to backend
*/
@@ -150,8 +200,7 @@ static void establish_persistent_connection(void)
{
int i;
BackendInfo *bkinfo;
- POOL_CONNECTION_POOL_SLOT *s;
-
+
for (i=0;i<NUM_BACKENDS;i++)
{
if (!VALID_BACKEND(i))
@@ -159,16 +208,21 @@ static void establish_persistent_connection(void)
if (slots[i] == NULL)
{
- bkinfo = pool_get_node_info(i);
- s = make_persistent_db_connection(bkinfo->backend_hostname,
+ PG_TRY();
+ {
+ bkinfo = pool_get_node_info(i);
+ slots[i] = make_persistent_db_connection(bkinfo->backend_hostname,
bkinfo->backend_port,
"postgres",
pool_config->sr_check_user,
pool_config->sr_check_password, true);
- if (s)
- slots[i] = s;
- else
+ }
+ PG_CATCH();
+ {
slots[i] = NULL;
+ }
+ PG_END_TRY();
+
}
}
}
@@ -231,10 +285,11 @@ static void check_replication_time_lag(void)
if (!slots[i])
{
- pool_debug("check_replication_time_lag: DB node is valid but no persistent connection");
- pool_error("check_replication_time_lag: could not connect to DB node %d, check sr_check_user and sr_check_password", i);
+ ereport(ERROR,
+ (errmsg("Failed to check replication time lag"),
+ errdetail("No persistent db connection for the node %d",i),
+ errhint("check sr_check_user and sr_check_password")));
- return;
}
if (PRIMARY_NODE_ID == i)
@@ -256,27 +311,32 @@ static void check_replication_time_lag(void)
}
if (!res)
{
- pool_error("check_replication_time_lag: %s result is null", query);
- return;
+ ereport(ERROR,
+ (errmsg("Failed to check replication time lag"),
+ errdetail("Query to node (%d) returned no result for node",i)));
}
if (res->numrows <= 0)
{
- pool_error("check_replication_time_lag: %s returns no rows", query);
free_select_result(res);
- return;
+ ereport(ERROR,
+ (errmsg("Failed to check replication time lag"),
+ errdetail("Query to node (%d) returned result with no rows",i)));
}
if (res->data[0] == NULL)
{
- pool_error("check_replication_time_lag: %s returns no data", query);
free_select_result(res);
- return;
+ ereport(ERROR,
+ (errmsg("Failed to check replication time lag"),
+ errdetail("Query to node (%d) returned no data",i)));
}
if (res->nullflags[0] == -1)
{
- pool_log("check_replication_time_lag: %s returns NULL", query);
free_select_result(res);
lsn[i] = 0;
+ ereport(ERROR,
+ (errmsg("Failed to check replication time lag"),
+ errdetail("Query to node (%d) returned NULL data",i)));
}
else
{
@@ -308,8 +368,9 @@ static void check_replication_time_lag(void)
!strcmp(pool_config->log_standby_delay, "if_over_threshold") &&
lag > pool_config->delay_threshold))
{
- pool_log("Replication of node:%d is behind %llu bytes from the primary server (node:%d)",
- i, lsn[PRIMARY_NODE_ID] - lsn[i], PRIMARY_NODE_ID);
+ ereport(LOG,
+ (errmsg("Replication of node:%d is behind %llu bytes from the primary server (node:%d)",
+ i, lsn[PRIMARY_NODE_ID] - lsn[i], PRIMARY_NODE_ID)));
}
}
}
@@ -332,8 +393,10 @@ static long text_to_lsn(char *text)
if (sscanf(text, "%X/%X", &xlogid, &xrecoff) != 2)
{
- pool_error("text_to_lsn: wrong log location format: %s", text);
- return 0;
+ ereport(ERROR,
+ (errmsg("Invalid lsn format"),
+ errdetail("wrong log location format: %s", text)));
+
}
lsn = xlogid * ((unsigned long long int)0xffffffff - WALSEGMENTSIZE) + xrecoff;
#ifdef DEBUG
diff --git a/src/utils/error/elog.c b/src/utils/error/elog.c
index dc457e386..a8c7766b2 100644
--- a/src/utils/error/elog.c
+++ b/src/utils/error/elog.c
@@ -271,6 +271,17 @@ errstart(int elevel, const char *filename, int lineno,
elevel = FATAL;
}
+ if(elevel == ERROR)
+ {
+ if(processType == PT_CHILD)
+ {
+ /*
+ * If the frontend connection exists, treat this error as FATAL
+ */
+ if(is_session_connected())
+ elevel = FATAL;
+ }
+ }
/*
* If the error level is ERROR or more, errfinish is not going to
* return to caller; therefore, if there is any stacked error already
diff --git a/src/utils/pool_stream.c b/src/utils/pool_stream.c
index ceb5d9008..d864929f9 100644
--- a/src/utils/pool_stream.c
+++ b/src/utils/pool_stream.c
@@ -39,6 +39,7 @@
#endif
#include "pool.h"
+#include "utils/elog.h"
#include "utils/pool_stream.h"
#include "pool_config.h"
@@ -120,6 +121,16 @@ void pool_close(POOL_CONNECTION *cp)
free(cp);
}
+void pool_read_with_error(POOL_CONNECTION *cp, void *buf, int len,
+ const char* err_context )
+{
+ if (pool_read(cp, buf, len) < 0)
+ {
+ ereport(ERROR,
+ (errmsg("failed to read data of length %d from DB node: %d",len,cp->db_node_id),
+ errdetail("error occurred when reading: %s",err_context?err_context:"")));
+ }
+}
/*
* read len bytes from cp
* returns 0 on success otherwise -1.
[parent: f84158903cff]