pgpool2.git / summary / log / commit / refs
commit 14ac0aead287ae67d0ef91463662d47e6ee1d9dc
Author: Tatsuo Ishii <ishii@postgresql.org>
Date: Sat May 19 16:30:49 2012 +0000
Fix on memory query cache does not work with extended query case.
(pgbench -M extended). To do this we need followings:
Add new parameter "bool cache_commit" to ReadyForQuery(). If
cache_commit is false, ReadyForQuery() does not do any cache commit
process. This is necessary for Execute() to call ReadyForQuery() to
process sending SYNC message in replication mode case.
Fix pool_fetch_from_memory_cache() to consume SYNC message after
sending "Command Complete" message.
Fix pool_process_query() to check all cache empty after calling
ProcessFrontendResponse(). This may or may not related to this
particular on memory query cache related fix though. Anyway it seems
we need the fix.
pool_memqcache.c | 19 +++++++++
pool_process_query.c | 117 +++++++++++++++++++++++++++++----------------------
pool_proto_modules.c | 24 ++++++++---
pool_proto_modules.h | 4 +-
4 files changed, 107 insertions(+), 57 deletions(-)
diff --git a/pool_memqcache.c b/pool_memqcache.c
index 61a527c5d..b28caad8c 100644
--- a/pool_memqcache.c
+++ b/pool_memqcache.c
@@ -572,6 +572,25 @@ POOL_STATUS pool_fetch_from_memory_cache(POOL_CONNECTION *frontend,
send_cached_messages(frontend, qcache, qcachelen);
free(qcache);
+ /*
+ * If we are doing extended query, wait and discard Sync
+ * message from frontend. This is neccessary to prevent
+ * receiving Sync message after Sending Ready for query.
+ */
+ if (pool_is_doing_extended_query_message())
+ {
+ char kind;
+ int32 len;
+
+ if (pool_flush(frontend))
+ return POOL_END;
+ if (pool_read(frontend, &kind, 1))
+ return POOL_END;
+ pool_debug("pool_fetch_from_memory_cache: expecting sync: %c", kind);
+ if (pool_read(frontend, &len, sizeof(len)))
+ return POOL_END;
+ }
+
/*
* send a "READY FOR QUERY"
*/
diff --git a/pool_process_query.c b/pool_process_query.c
index 3a82c393f..adc3734b7 100644
--- a/pool_process_query.c
+++ b/pool_process_query.c
@@ -232,74 +232,89 @@ POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
}
else
{
- /* Ok, query is not in progress. To make sure that we
- * have any pending data in backends. */
-
- /* If we have pending data in master, we need to process it */
- if (!pool_ssl_pending(MASTER(backend)) &&
- !pool_read_buffer_is_empty(MASTER(backend)))
+ /* Ok, query is not in progress.
+ * ProcessFrontendResponse() may consume all pending
+ * data. Check if we have any pending data. If no,
+ * call read_packets_and_process() and wait for data
+ * arrival.
+ */
+ if (is_cache_empty(frontend, backend))
{
- status = ProcessBackendResponse(frontend, backend, &state, &num_fields);
+ bool cont = true;
+ status = read_packets_and_process(frontend, backend, reset_request, &state, &num_fields, &cont);
if (status != POOL_CONTINUE)
return status;
+ else if (!cont) /* Detected admin shutdown */
+ return status;
}
else
{
- for (i=0;i<NUM_BACKENDS;i++)
+ /* If we have pending data in master, we need to process it */
+ if (!pool_ssl_pending(MASTER(backend)) &&
+ !pool_read_buffer_is_empty(MASTER(backend)))
{
- if (!VALID_BACKEND(i))
- continue;
-
- if (pool_ssl_pending(CONNECTION(backend, i)) ||
- pool_read_buffer_is_empty(CONNECTION(backend, i)))
+ status = ProcessBackendResponse(frontend, backend, &state, &num_fields);
+ if (status != POOL_CONTINUE)
+ return status;
+ }
+ else
+ {
+ for (i=0;i<NUM_BACKENDS;i++)
{
- /* If we have pending data in master, we need to process it */
- if (IS_MASTER_NODE_ID(i))
- {
- status = ProcessBackendResponse(frontend, backend, &state, &num_fields);
- if (status != POOL_CONTINUE)
- return status;
- break;
- }
- else
+ if (!VALID_BACKEND(i))
+ continue;
+
+ if (pool_ssl_pending(CONNECTION(backend, i)) ||
+ pool_read_buffer_is_empty(CONNECTION(backend, i)))
{
- char kind;
- int len;
- char *string;
-
- /* 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)
+ /* If we have pending data in master, we need to process it */
+ if (IS_MASTER_NODE_ID(i))
{
- pool_error("pool_process_query: error while reading message kind from backend %d", i);
- return POOL_END;
+ status = ProcessBackendResponse(frontend, backend, &state, &num_fields);
+ if (status != POOL_CONTINUE)
+ return status;
+ break;
}
- pool_log("pool_process_query: discard %c packet from backend %d", kind, i);
-
- if (MAJOR(backend) == PROTO_MAJOR_V3)
+ else
{
- if (pool_read(CONNECTION(backend, i), &len, sizeof(len)) < 0)
+ char kind;
+ int len;
+ char *string;
+
+ /* 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 length from backend %d", i);
+ pool_error("pool_process_query: error while reading message kind from backend %d", i);
return POOL_END;
}
- len = ntohl(len) - 4;
- string = pool_read2(CONNECTION(backend, i), len);
- if (string == NULL)
+ pool_log("pool_process_query: discard %c packet from backend %d", kind, i);
+
+ if (MAJOR(backend) == PROTO_MAJOR_V3)
{
- pool_error("pool_process_query: error while reading rest of message from backend %d", i);
- return POOL_END;
+ 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_END;
+ }
+ len = ntohl(len) - 4;
+ string = pool_read2(CONNECTION(backend, i), len);
+ if (string == NULL)
+ {
+ pool_error("pool_process_query: error while reading rest of message from backend %d", i);
+ return POOL_END;
+ }
}
- }
- else
- {
- string = pool_read_string(CONNECTION(backend, i), &len, 0);
- if (string == NULL)
+ else
{
- pool_error("pool_process_query: error while reading rest of message from backend %d", i);
- return POOL_END;
+ string = pool_read_string(CONNECTION(backend, i), &len, 0);
+ if (string == NULL)
+ {
+ pool_error("pool_process_query: error while reading rest of message from backend %d", i);
+ return POOL_END;
+ }
}
}
}
@@ -3358,6 +3373,8 @@ POOL_STATUS read_kind_from_backend(POOL_CONNECTION *frontend, POOL_CONNECTION_PO
return POOL_ERROR;
}
+ pool_debug("read_kind_from_backend: kind: %c from %d th backend", kind, i);
+
/*
* Read and discard parameter status
*/
diff --git a/pool_proto_modules.c b/pool_proto_modules.c
index a076e3467..65d813e38 100644
--- a/pool_proto_modules.c
+++ b/pool_proto_modules.c
@@ -734,6 +734,7 @@ POOL_STATUS Execute(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
pool_ps_idle_display(backend);
pool_set_skip_reading_from_backends();
pool_stats_count_up_num_cache_hits();
+ pool_unset_query_in_progress();
return POOL_CONTINUE;
}
}
@@ -1014,6 +1015,9 @@ POOL_STATUS Parse(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
pool_memory_context_switch_to(old_context);
+ /*
+ * If in replication mode, send "SYNC" message if not in a transaction.
+ */
if (REPLICATION)
{
char kind;
@@ -1039,7 +1043,10 @@ POOL_STATUS Parse(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
return POOL_END;
}
- if (ReadyForQuery(frontend, backend, 0) != POOL_CONTINUE)
+ /*
+ * SYNC message returns "Ready for Query" message.
+ */
+ if (ReadyForQuery(frontend, backend, 0, false) != POOL_CONTINUE)
{
pool_query_context_destroy(query_context);
return POOL_END;
@@ -1416,13 +1423,16 @@ POOL_STATUS FunctionCall3(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backe
/*
* Process ReadyForQuery('Z') message.
+ * If send_ready is true, send 'Z' message to frontend.
+ * If cache_commit is true, commit or discard query cache according to
+ * transaction state.
*
* - if the error status "mismatch_ntuples" is set, send an error query
* to all DB nodes to abort transaction or do failover.
* - internal transaction is closed
*/
POOL_STATUS ReadyForQuery(POOL_CONNECTION *frontend,
- POOL_CONNECTION_POOL *backend, int send_ready)
+ POOL_CONNECTION_POOL *backend, bool send_ready, bool cache_commit)
{
int i;
int len;
@@ -1669,7 +1679,7 @@ POOL_STATUS ReadyForQuery(POOL_CONNECTION *frontend,
}
/* Memory cache enabled? */
- if (pool_config->memory_cache_enabled)
+ if (cache_commit && pool_config->memory_cache_enabled)
{
if (pool_is_doing_extended_query_message())
{
@@ -1700,6 +1710,10 @@ POOL_STATUS ReadyForQuery(POOL_CONNECTION *frontend,
pool_unset_query_in_progress();
}
+ /*
+ * If we ared doing extended query, we cannot destroy query context now.
+ * We postpone it until the portal is deleted.
+ */
if (session_context->query_context && !pool_is_doing_extended_query_message())
{
if (!(node && IsA(node, PrepareStmt)))
@@ -2478,7 +2492,7 @@ POOL_STATUS ProcessBackendResponse(POOL_CONNECTION *frontend,
break;
case 'Z': /* ReadyForQuery */
- status = ReadyForQuery(frontend, backend, 1);
+ status = ReadyForQuery(frontend, backend, true, true);
pool_debug("ProcessBackendResponse: Ready For Query");
break;
@@ -2607,7 +2621,7 @@ POOL_STATUS ProcessBackendResponse(POOL_CONNECTION *frontend,
break;
case 'Z': /* ReadyForQuery */
- status = ReadyForQuery(frontend, backend, 1);
+ status = ReadyForQuery(frontend, backend, true, true);
break;
default:
diff --git a/pool_proto_modules.h b/pool_proto_modules.h
index c5b691575..68053b9c8 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-2011 PgPool Global Development Group
+ * Copyright (c) 2003-2012 PgPool Global Development Group
*
* Permission to use, copy, modify, and distribute this software and
* its documentation for any purpose and without fee is hereby
@@ -82,7 +82,7 @@ extern POOL_STATUS FunctionCall3(POOL_CONNECTION *frontend,
int len, char *contents);
extern POOL_STATUS ReadyForQuery(POOL_CONNECTION *frontend,
- POOL_CONNECTION_POOL *backend, int send_ready);
+ POOL_CONNECTION_POOL *backend, bool send_ready, bool cache_commit);
extern POOL_STATUS ParseComplete(POOL_CONNECTION *frontend,
POOL_CONNECTION_POOL *backend);
[parent: 89a2e624aaa6]