pgpool2.git / summary / log / commit / refs
commit bb295a22ad47001758b0b38968f53c776f439f2f
Author: Tatsuo Ishii <ishii@postgresql.org>
Date: Thu Mar 03 06:12:06 2016 +0000
Fix query stack problems in extended protocol case.
This fix is for both bug167 and 168.
The bug 167 iss caused by following message sequence:
13:38:24.947 (1) FE=> Parse(stmt=null,query="BEGIN",oids={})
13:38:24.947 (1) FE=> Bind(stmt=null,portal=null)
13:38:24.947 (1) FE=> Execute(portal=null,limit=0)
13:38:24.948 (1) FE=> Parse(stmt=null,query="SELECT t.typname,t.oid FROM pg_catalog.pg_type t JOIN pg_catalog.pg_namespace n ON (t.typnamespace = n.oid) WHERE n.nspname != 'pg_toast'",oids={})
13:38:24.948 (1) FE=> Bind(stmt=null,portal=null)
13:38:24.948 (1) FE=> Describe(portal=null)
13:38:24.948 (1) FE=> Execute(portal=null,limit=0)
13:38:24.948 (1) FE=> Sync
As you can see before sync for "BEGIN" is issued next parse message
arrives. Then sync is arrived after the second query. Since the query
in progress flag is unset, pgpool-II tries to read data from backend 0
(this is ok because the messages have been sent to it) and backend 1
which will block since no messages are sent in this case (load balance
node = 0).
Fix is, decide which node to read by looking at sync map. For this
purpose, pool_read_message_length(), ReadyForQuery(),
SimpleForwardToFrontend(), read_kind_from_backend() are modified to
use new function pool_use_sync_map().
Another important fix is around line 211 of
pool_process_query(). Before we call ProcessBackendResponse() only
when query is in progress. This is wrong since there might be a
pending data in the backend cache and we need to process it. This
could happen if do_query() stacks up pending data. Otherwise pgpool-II
repeats to visit the line infinitely. I suspect this is the cause
reported in pgpool-hackers: 1432. New function
is_backend_cache_empty() is added to check if there's pending data in
the backend cache in ProcessBackendResponse().
src/auth/pool_auth.c | 14 ++++++-
src/context/pool_query_context.c | 10 ++---
src/context/pool_session_context.c | 28 +++++++++++++
src/include/context/pool_session_context.h | 12 +++++-
src/include/pool.h | 2 +
src/protocol/pool_process_query.c | 63 ++++++++++++++++++++++--------
src/protocol/pool_proto_modules.c | 20 +++++++++-
7 files changed, 123 insertions(+), 26 deletions(-)
diff --git a/src/auth/pool_auth.c b/src/auth/pool_auth.c
index 597d56bdf..2f259df30 100644
--- a/src/auth/pool_auth.c
+++ b/src/auth/pool_auth.c
@@ -5,7 +5,7 @@
* pgpool: a language independent connection pool server for PostgreSQL
* written by Tatsuo Ishii
*
- * Copyright (c) 2003-2015 PgPool Global Development Group
+ * Copyright (c) 2003-2016 PgPool Global Development Group
*
* Permission to use, copy, modify, and distribute this software and
* its documentation for any purpose and without fee is hereby
@@ -23,6 +23,7 @@
*/
#include "pool.h"
+#include "context/pool_session_context.h"
#include "utils/pool_stream.h"
#include "pool_config.h"
#include "auth/pool_passwd.h"
@@ -1092,6 +1093,10 @@ int pool_read_message_length(POOL_CONNECTION_POOL *cp)
{
int length, length0;
int i;
+ POOL_SYNC_MAP_STATE use_sync_map = pool_use_sync_map();
+
+ /* Check whether we can use the sync map or not */
+ use_sync_map = pool_use_sync_map();
/* read message from master node */
pool_read(CONNECTION(cp, MASTER_NODE_ID), &length0, sizeof(length0));
@@ -1103,7 +1108,12 @@ int pool_read_message_length(POOL_CONNECTION_POOL *cp)
for (i=0;i<NUM_BACKENDS;i++)
{
- if (!VALID_BACKEND(i) || IS_MASTER_NODE_ID(i))
+ if (!VALID_BACKEND(i) || IS_MASTER_NODE_ID(i) || use_sync_map == POOL_SYNC_MAP_EMPTY)
+ {
+ continue;
+ }
+
+ if (use_sync_map == POOL_SYNC_MAP_IS_VALID && !pool_is_set_sync_map(i))
{
continue;
}
diff --git a/src/context/pool_query_context.c b/src/context/pool_query_context.c
index 6a631b70f..c24016224 100644
--- a/src/context/pool_query_context.c
+++ b/src/context/pool_query_context.c
@@ -848,15 +848,15 @@ POOL_STATUS pool_extended_send_and_wait(POOL_QUERY_CONTEXT *query_context,
per_node_statement_log(backend, i, msgbuf);
}
+ /* Set sync map so that we could wait for response from appropreate node */
+ pool_set_sync_map(i);
+ ereport(DEBUG1,
+ (errmsg("pool_send_and_wait: pool_set_sync_map: %d", i)));
+
/* if Execute message, count up stats count */
if (*kind == 'E')
{
stat_count_up(i, query_context->parse_tree);
-
- /* Set sync map so that we could wait for response from appropreate node */
- pool_set_sync_map(i);
- ereport(DEBUG1,
- (errmsg("pool_send_and_wait: pool_set_sync_map: %d", i)));
}
send_extended_protocol_message(backend, i, kind, str_len, str);
diff --git a/src/context/pool_session_context.c b/src/context/pool_session_context.c
index 5221577c6..8fd95a884 100644
--- a/src/context/pool_session_context.c
+++ b/src/context/pool_session_context.c
@@ -929,6 +929,34 @@ void pool_clear_sync_map(void)
memset(&session_context->sync_map, 0, sizeof(session_context->sync_map));
}
+/*
+ * Check whether we should 1) ignore sync map, 2) consult sync map, or 3) we
+ * cannot use sync map because it's empty.
+ */
+bool pool_use_sync_map(void)
+{
+ int i;
+
+ if (!session_context)
+ return POOL_IGNORE_SYNC_MAP;
+
+ if (STREAM && !pool_is_query_in_progress() && pool_is_doing_extended_query_message())
+ {
+ for (i=0;i<NUM_BACKENDS;i++)
+ {
+ if (pool_is_set_sync_map(i))
+ {
+ ereport(DEBUG1,
+ (errmsg("pool_use_sync_map: set use sync map: %d", i)));
+ return POOL_SYNC_MAP_IS_VALID; /* yes, we can use sync map */
+ }
+ }
+ return POOL_SYNC_MAP_EMPTY; /* no, we cannot use sync map */
+ }
+
+ return POOL_IGNORE_SYNC_MAP;
+}
+
/*
* Set pending response
*/
diff --git a/src/include/context/pool_session_context.h b/src/include/context/pool_session_context.h
index 82512150d..5a4486540 100644
--- a/src/include/context/pool_session_context.h
+++ b/src/include/context/pool_session_context.h
@@ -6,7 +6,7 @@
* pgpool: a language independent connection pool server for PostgreSQL
* written by Tatsuo Ishii
*
- * Copyright (c) 2003-2011 PgPool Global Development Group
+ * Copyright (c) 2003-2016 PgPool Global Development Group
*
* Permission to use, copy, modify, and distribute this software and
* its documentation for any purpose and without fee is hereby
@@ -44,6 +44,15 @@ typedef enum {
POOL_SERIALIZABLE /* Serializable */
} POOL_TRANSACTION_ISOLATION;
+/*
+ * Return values for pool_use_sync_map
+ */
+typedef enum {
+ POOL_IGNORE_SYNC_MAP, /* Please ignore sync map */
+ POOL_SYNC_MAP_IS_VALID, /* Sync map is valid */
+ POOL_SYNC_MAP_EMPTY, /* Sync map is empty */
+} POOL_SYNC_MAP_STATE;
+
/*
* Message content of extended query
*/
@@ -246,6 +255,7 @@ extern void pool_set_sync_map(int node_id);
extern bool pool_is_set_sync_map(int node_id);
extern int pool_get_nth_sync_map(int nth);
extern void pool_clear_sync_map(void);
+extern bool pool_use_sync_map(void);
extern void pool_set_pending_response(void);
extern void pool_unset_pending_response(void);
extern bool pool_is_pending_response(void);
diff --git a/src/include/pool.h b/src/include/pool.h
index 5c6202bda..ee8b1109f 100644
--- a/src/include/pool.h
+++ b/src/include/pool.h
@@ -517,6 +517,8 @@ extern POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
extern int pool_do_auth(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend);
extern int pool_do_reauth(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *cp);
+extern bool is_backend_cache_empty(POOL_CONNECTION_POOL *backend);
+
/* SSL functionality */
extern void pool_ssl_negotiate_serverclient(POOL_CONNECTION *cp);
extern void pool_ssl_negotiate_clientserver(POOL_CONNECTION *cp);
diff --git a/src/protocol/pool_process_query.c b/src/protocol/pool_process_query.c
index a6eb35286..fd3b76ffa 100644
--- a/src/protocol/pool_process_query.c
+++ b/src/protocol/pool_process_query.c
@@ -191,7 +191,7 @@ POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
/*
* If we are not processing a query, now is the time to
- * extract retrieve pending data from buffer stack if any.
+ * extract pending data from buffer stack if any.
*/
if (!pool_is_query_in_progress())
{
@@ -204,9 +204,11 @@ POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
}
/*
- * If we are prcessing query, process it.
+ * If we are processing query, process it. Even if we are not
+ * processing query, process backend response if there's pending data
+ * in backend cache.
*/
- if (pool_is_query_in_progress())
+ if (pool_is_query_in_progress() || !is_backend_cache_empty(backend))
{
status = ProcessBackendResponse(frontend, backend, &state, &num_fields);
if (status != POOL_CONTINUE)
@@ -712,6 +714,7 @@ POOL_STATUS SimpleForwardToFrontend(char kind, POOL_CONNECTION *frontend,
char *p1 = NULL;
int sendlen;
int i;
+ POOL_SYNC_MAP_STATE use_sync_map = pool_use_sync_map();
#ifdef NOT_USED
/*
@@ -748,8 +751,16 @@ POOL_STATUS SimpleForwardToFrontend(char kind, POOL_CONNECTION *frontend,
{
for (i=0;i<NUM_BACKENDS;i++)
{
+ if (use_sync_map == POOL_SYNC_MAP_EMPTY)
+ continue;
+
if (VALID_BACKEND(i) && !IS_MASTER_NODE_ID(i))
{
+ if (use_sync_map == POOL_SYNC_MAP_IS_VALID && !pool_is_set_sync_map(i))
+ {
+ continue;
+ }
+
pool_read(CONNECTION(backend, i), &len, sizeof(len));
len = ntohl(len);
@@ -3001,22 +3012,12 @@ int is_drop_database(Node *node)
}
/*
- * check if any pending data remains.
+ * check if any pending data remains in backend.
*/
-static bool is_cache_empty(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend)
+bool is_backend_cache_empty(POOL_CONNECTION_POOL *backend)
{
int i;
- /*
- * If SSL is enabled, we need to check SSL internal buffer
- * is empty or not first.
- */
- if (pool_ssl_pending(frontend))
- return false;
-
- if (!pool_read_buffer_is_empty(frontend))
- return false;
-
for (i=0;i<NUM_BACKENDS;i++)
{
if (!VALID_BACKEND(i))
@@ -3036,6 +3037,24 @@ static bool is_cache_empty(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *back
return true;
}
+/*
+ * check if any pending data remains.
+*/
+static bool is_cache_empty(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend)
+{
+ /*
+ * If SSL is enabled, we need to check SSL internal buffer
+ * is empty or not first.
+ */
+ if (pool_ssl_pending(frontend))
+ return false;
+
+ if (!pool_read_buffer_is_empty(frontend))
+ return false;
+
+ return is_backend_cache_empty(backend);
+}
+
/*
* check if query is needed to wait completion
*/
@@ -3178,9 +3197,10 @@ void read_kind_from_backend(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *bac
double max_count = 0;
int degenerate_node_num = 0; /* number of backends degeneration requested */
int degenerate_node[MAX_NUM_BACKENDS]; /* degeneration requested backend list */
-
+ bool doing_extended_message = false; /* are we doing extended protocol? */
POOL_SESSION_CONTEXT *session_context = pool_get_session_context(false);
POOL_QUERY_CONTEXT *query_context = session_context->query_context;
+ POOL_SYNC_MAP_STATE use_sync_map = pool_use_sync_map();
int num_executed_nodes = 0;
int first_node = -1;
@@ -3218,8 +3238,19 @@ void read_kind_from_backend(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *bac
/* initialize degenerate record */
degenerate_node[i] = 0;
+ if (!VALID_BACKEND(i) || use_sync_map == POOL_SYNC_MAP_EMPTY)
+ {
+ kind_list[i] = 0;
+ continue;
+ }
+
if (VALID_BACKEND(i))
{
+ if (use_sync_map == POOL_SYNC_MAP_IS_VALID && !pool_is_set_sync_map(i))
+ {
+ continue;
+ }
+
num_executed_nodes++;
if (first_node < 0)
diff --git a/src/protocol/pool_proto_modules.c b/src/protocol/pool_proto_modules.c
index 756bb2ff8..8c2deb403 100644
--- a/src/protocol/pool_proto_modules.c
+++ b/src/protocol/pool_proto_modules.c
@@ -1419,9 +1419,11 @@ POOL_STATUS ReadyForQuery(POOL_CONNECTION *frontend,
POOL_SESSION_CONTEXT *session_context;
Node *node = NULL;
char *query = NULL;
+ POOL_SYNC_MAP_STATE use_sync_map;
/* Get session context */
session_context = pool_get_session_context(false);
+ use_sync_map = pool_use_sync_map();
/*
* If the numbers of update tuples are differ and
@@ -1559,9 +1561,14 @@ POOL_STATUS ReadyForQuery(POOL_CONNECTION *frontend,
for (i=0;i<NUM_BACKENDS;i++)
{
- if (!VALID_BACKEND(i))
+ if (!VALID_BACKEND(i) || use_sync_map == POOL_SYNC_MAP_EMPTY)
continue;
+ if (use_sync_map == POOL_SYNC_MAP_IS_VALID && !pool_is_set_sync_map(i))
+ {
+ continue;
+ }
+
if (pool_read(CONNECTION(backend, i), &kind, sizeof(kind)))
return POOL_END;
@@ -2349,7 +2356,16 @@ POOL_STATUS ProcessBackendResponse(POOL_CONNECTION *frontend,
{
if (pool_is_query_in_progress())
pool_unset_query_in_progress();
- return POOL_CONTINUE;
+
+ /*
+ * Check if If we have pending data in backend connection cache. If we
+ * do, it is likely that a sync message has been sent to backend and the
+ * backend replied back to us. So we need to process it.
+ */
+ if (is_backend_cache_empty(backend))
+ {
+ return POOL_CONTINUE;
+ }
}
if (pool_is_skip_reading_from_backends())
[parent: 63a0b7042b4b]