pgpool2.git / summary / log / commit / refs
commit 619beb0e2c274b24f6b8fedc841bf418f45f9d0f
Author: Tatsuo Ishii <ishii at sraoss.co.jp>
Date: Mon Jul 20 12:21:54 2009 +0000
Check frontend connection while waiting for response from backend.
This is done in wait_for_query_response() by writing a dummy parameter
status packet to pgpool client every 1 second. If the write fails, a
cancel packet is sent to backend to free backend.
This is usefull to prevent pgpool child from waiting forever for
backend response. Pgpool client will diconnect to pgpool if the query
takes too long. This is typical use case for Web applications.
child.c | 3 +--
pool.h | 1 +
pool_process_query.c | 75 +++++++++++++++++++++++++++++++++++++++-------------
pool_proto_modules.c | 40 ++++++++++++++++++++++------
pool_proto_modules.h | 4 +--
5 files changed, 93 insertions(+), 30 deletions(-)
diff --git a/child.c b/child.c
index c907c19fb..e19c07831 100644
--- a/child.c
+++ b/child.c
@@ -60,7 +60,6 @@
static POOL_CONNECTION *do_accept(int unix_fd, int inet_fd, struct timeval *timeout);
static StartupPacket *read_startup_packet(POOL_CONNECTION *cp);
static POOL_CONNECTION_POOL *connect_backend(StartupPacket *sp, POOL_CONNECTION *frontend);
-static void cancel_request(CancelPacket *sp);
static RETSIGTYPE die(int sig);
static RETSIGTYPE close_idle_connection(int sig);
static RETSIGTYPE wakeup_handler(int sig);
@@ -1079,7 +1078,7 @@ int send_startup_packet(POOL_CONNECTION_POOL_SLOT *cp)
/*
* process cancel request
*/
-static void cancel_request(CancelPacket *sp)
+void cancel_request(CancelPacket *sp)
{
int len;
int fd;
diff --git a/pool.h b/pool.h
index 57b767e63..576b57069 100644
--- a/pool.h
+++ b/pool.h
@@ -583,5 +583,6 @@ extern void finish_recovery(void);
/* child.c */
extern void pool_set_nonblock(int fd);
extern void pool_unset_nonblock(int fd);
+extern void cancel_request(CancelPacket *sp);
#endif /* POOL_H */
diff --git a/pool_process_query.c b/pool_process_query.c
index 5c46300d7..3804c0c4f 100644
--- a/pool_process_query.c
+++ b/pool_process_query.c
@@ -819,16 +819,55 @@ POOL_STATUS send_simplequery_message(POOL_CONNECTION *backend, int len, char *st
}
/*
- * wait for query response from a single node.
+ * Wait for query response from single node. This checks frontend
+ * connection by writing dummy parameter status packet every 1
+ * seccond, and if the connection broke, returns error since there's
+ * no point in that waiting until backend returns response.
*/
-POOL_STATUS wait_for_query_response(POOL_CONNECTION *backend, char *string)
+POOL_STATUS wait_for_query_response(POOL_CONNECTION *frontend, POOL_CONNECTION *backend, char *string)
{
- /*
- * we need to wait for backend completing the query.
- */
- pool_debug("waiting for backend %d completing the query", backend->db_node_id);
- if (synchronize(backend))
- return POOL_END;
+#define DUMMY_PARAMETER "pgpool_dummy_param"
+#define DUMMY_VALUE "pgpool_dummy_value"
+
+ int status;
+ int plen;
+
+ pool_debug("wait_for_query_response: waiting for backend %d completing the query", backend->db_node_id);
+
+ for (;;)
+ {
+ /* Check to see if data from backend is ready */
+ pool_set_timeout(1);
+ status = pool_check_fd(backend);
+ pool_set_timeout(0);
+
+ if (status < 0) /* error ? */
+ {
+ pool_error("wait_for_query_response: backend error occured while waiting for backend response");
+ return POOL_END;
+ } else if (status > 0) /* data is not ready */
+ {
+ /* Write dummy parameter staus packet to check if the socket to frontend is ok */
+ if (pool_write(frontend, "S", 1) < 0)
+ return POOL_END;
+ plen = sizeof(DUMMY_PARAMETER)+sizeof(DUMMY_VALUE)+sizeof(plen);
+ plen = htonl(plen);
+ if (pool_write(frontend, &plen, sizeof(plen)) < 0)
+ return POOL_END;
+ if (pool_write(frontend, DUMMY_PARAMETER, sizeof(DUMMY_PARAMETER)) < 0)
+ return POOL_END;
+ if (pool_write(frontend, DUMMY_VALUE, sizeof(DUMMY_VALUE)) < 0)
+ return POOL_END;
+ if (pool_flush_it(frontend) < 0)
+ {
+ pool_error("wait_for_query_response: frontend error ouccured while waiting for backend reply");
+ return POOL_END;
+ }
+ }
+ else
+ break;
+
+ }
return POOL_CONTINUE;
}
@@ -891,7 +930,8 @@ void pool_set_timeout(int timeoutval)
}
/*
- * wait until read data is ready
+ * Wait until read data is ready.
+ * return values: 0: normal 1: data is not ready -1: error
*/
int pool_check_fd(POOL_CONNECTION *cp)
{
@@ -930,7 +970,7 @@ int pool_check_fd(POOL_CONNECTION *cp)
break;
}
else if (fds == 0) /* timeout */
- break;
+ return 1;
if (FD_ISSET(fd, &exceptmask))
{
@@ -1699,14 +1739,12 @@ POOL_STATUS SimpleForwardToFrontend(char kind, POOL_CONNECTION *frontend, POOL_C
free(p1);
if (status)
return POOL_END;
-#ifdef NOT_USED
+
if (kind == 'A') /* notification response */
{
pool_flush(frontend); /* we need to immediately notice to frontend */
}
else if (kind == 'E') /* error response? */
-#endif
- if (kind == 'E') /* error response? */
{
int i;
int res1;
@@ -2097,17 +2135,17 @@ static int reset_backend(POOL_CONNECTION_POOL *backend, int qcnt)
/* if transaction state is idle, we don't need to issue ABORT */
if (TSTATE(backend) == 'I' && !strcmp("ABORT", query))
+ return 0;
+
pool_set_timeout(10);
- return 0;
+ if (SimpleQuery(NULL, backend, query) != POOL_CONTINUE)
{
pool_set_timeout(0);
-
- }
- if (SimpleQuery(NULL, backend, query) != POOL_CONTINUE)
- pool_set_timeout(0);
return -1;
+ }
+ pool_set_timeout(0);
return 1;
}
@@ -4115,6 +4153,7 @@ POOL_STATUS end_internal_transaction(POOL_CONNECTION_POOL *backend)
{
if (VALID_BACKEND(i) && !IS_MASTER_NODE_ID(i))
{
+ /* COMMIT success? */
if (do_command(CONNECTION(backend, i), "COMMIT", MAJOR(backend), 1) != POOL_CONTINUE)
{
internal_transaction_started = 0;
diff --git a/pool_proto_modules.c b/pool_proto_modules.c
index a7e32b22f..4121897b0 100644
--- a/pool_proto_modules.c
+++ b/pool_proto_modules.c
@@ -514,18 +514,32 @@ POOL_STATUS NotificationResponse(POOL_CONNECTION *frontend,
if (REPLICATION || PARALLEL_MODE)
{
- /* check if query is "COMMIT" */
+ /* check if query is "COMMIT" or "ROLLBACK" */
commit = is_commit_query(node);
free_parser();
- /* send query to master node */
+ /*
+ * Query is not commit/rollback
+ */
if (!commit)
{
+ /* Send the query to master node */
+
if (send_simplequery_message(MASTER(backend), len, string, MAJOR(backend)) != POOL_CONTINUE)
return POOL_END;
- if (wait_for_query_response(MASTER(backend), string) != POOL_CONTINUE)
+ if (wait_for_query_response(frontend, MASTER(backend), string) != POOL_CONTINUE)
+ {
+ /* Cancel current transaction */
+ CancelPacket cancel_packet;
+
+ cancel_packet.protoVersion = htonl(80877102);
+ cancel_packet.pid = MASTER_CONNECTION(backend)->pid;
+ cancel_packet.key= MASTER_CONNECTION(backend)->key;
+ cancel_request(&cancel_packet);
+
return POOL_END;
+ }
/*
* Check dead lock error on the master node and abort
@@ -607,17 +621,17 @@ POOL_STATUS NotificationResponse(POOL_CONNECTION *frontend,
if (!VALID_BACKEND(i) || IS_MASTER_NODE_ID(i))
continue;
- if (wait_for_query_response(CONNECTION(backend, i), string) != POOL_CONTINUE)
+ if (wait_for_query_response(frontend, CONNECTION(backend, i), string) != POOL_CONTINUE)
return POOL_END;
}
- /* send "COMMIT" to master node if query is "COMMIT" */
+ /* send "COMMIT" or "ROLLBACK" to only master node if query is "COMMIT" or "ROLLBACK" */
if (commit)
{
if (send_simplequery_message(MASTER(backend), len, string, MAJOR(backend)) != POOL_CONTINUE)
return POOL_END;
- if (wait_for_query_response(MASTER(backend), string) != POOL_CONTINUE)
+ if (wait_for_query_response(frontend, MASTER(backend), string) != POOL_CONTINUE)
return POOL_END;
TSTATE(backend) = 'I';
@@ -629,8 +643,18 @@ POOL_STATUS NotificationResponse(POOL_CONNECTION *frontend,
if (send_simplequery_message(MASTER(backend), len, string, MAJOR(backend)) != POOL_CONTINUE)
return POOL_END;
- if (wait_for_query_response(MASTER(backend), string) != POOL_CONTINUE)
- return POOL_END;
+ if (wait_for_query_response(frontend, MASTER(backend), string) != POOL_CONTINUE)
+ {
+ /* Cancel current transaction */
+ CancelPacket cancel_packet;
+
+ cancel_packet.protoVersion = htonl(80877102);
+ cancel_packet.pid = MASTER_CONNECTION(backend)->pid;
+ cancel_packet.key= MASTER_CONNECTION(backend)->key;
+ cancel_request(&cancel_packet);
+
+ return POOL_END;
+ }
}
return POOL_CONTINUE;
diff --git a/pool_proto_modules.h b/pool_proto_modules.h
index 0888daa94..2e444e98b 100644
--- a/pool_proto_modules.h
+++ b/pool_proto_modules.h
@@ -6,7 +6,7 @@
* pgpool: a language independent connection pool server for PostgreSQL
* written by Tatsuo Ishii
*
- * Copyright (c) 2003-2008 PgPool Global Development Group
+ * Copyright (c) 2003-2009 PgPool Global Development Group
*
* Permission to use, copy, modify, and distribute this software and
* its documentation for any purpose and without fee is hereby
@@ -141,7 +141,7 @@ extern POOL_STATUS ProcessFrontendResponse(POOL_CONNECTION *frontend,
POOL_CONNECTION_POOL *backend);
-extern POOL_STATUS wait_for_query_response(POOL_CONNECTION *backend, char *string);
+extern POOL_STATUS wait_for_query_response(POOL_CONNECTION *frontend, POOL_CONNECTION *backend, char *string);
extern int is_select_query(Node *node, char *sql);
extern int is_sequence_query(Node *node);
extern int is_start_transaction_query(Node *node);
[parent: 5e0048e144c3]