pgpool2.git / summary / log / commit / refs
commit e7b4de8d457e223234975b4ceaf8c345acdc2492
Author: Tatsuo Ishii <ishii@postgresql.org>
Date: Thu Sep 24 02:08:17 2015 +0000
Mega patch to enhance performance in extended protocol mode.
For now it only affects to streaming replication mode, that is, the
performance of other modes remains same. In the future we may be able
to enhance this.
Anyway, basic idea here is, removing "flush" message which used to be
sent in each step of extended protocol messages (parse, bind, describe
and execute). This brought significant communication overhead. To
achieve the goal following modifications have been made.
- New data structure called "pending message queue" is created to
manage which response of particular message is pending. When
Parse/Bind/Close message are received, each message is en-queued.
The information is used to process those response messages, when
Parse complete/Bind completes and Close compete message are received
because they don't have any information regarding statement/portal.
- New state variable to represent that there's any "pending response"
exists. If an extended protocol message, for example, parse is sent
and still the response to parse (i.e. parse complete) is not yet
received, the variable is set to true.
- If there's any pending response, do_query() sends flush message to
retrieve any pending data from primary backend and save it. After
finishing the job, the saved data is restored. This will keep the
original response message sequence.
- Deal with a special case in read_kind_from_backend(). Certain client
(at least JDBC driver has this habit) does not send a sync message
after an execute message. See the comment in
read_kind_from_backend() for more details.
- New data structure called "sync map" to manage sync message response
is added. Used for managing sync message response in streaming
replication mode. In streaming replication mode, there are at most
two nodes involved. One is always primary, and the other (if any)
could be standby. If the load balancing manager chooses the primary
as the load balancing node, only the primary node is involved. Every
time extended protocol is sent to backend, we set true on the member
of array below. It is cleared after sync message is processed and
command complete or error response message is processed.
- New parameter "bool nowait" is added to
pool_extended_send_and_wait() to control whether flush message is
sent or not after sending extended protocol command.
- Move CommandComplete() to new file src/protocol/CommandComplete.c to
make maintainer's life easier.
Initial performance evaluation:
- Hard ware: Let's note CF-SX3, mem 16GB, SSD 512GB, Core i7-4600U (2 cores)
- OS: Ubuntu 14.04 LTS
- 2 PostgreSQL streaming replication setup using pgpool_setup
- pgbench options: pgbench -M extended -c 16 -j 8 -S -T 30 test (scale=1)
- Results (excluding connections establishing):
pgpool-II 3.4.4: 7274.926104 TPS, this commit: 11290.026468 TPS
So, this committed version is 55.2% faster than 3.4.3.
TODOs:
- parse_before_bind() in pool_proto_modules is ifdef'ed out. Even
without this, all regression tests are passed. But I am not sure if
it's really harmless.
- When parse message is sent, it is possible that deadlock occurs
according to the comment. I'm not sure if it's still true or not. If
true, we need to fix it.
src/Makefile.am | 3 +-
src/Makefile.in | 7 +-
src/context/pool_query_context.c | 82 +++--
src/context/pool_session_context.c | 295 +++++++++++++++-
src/include/context/pool_query_context.h | 2 +-
src/include/context/pool_session_context.h | 63 ++++
src/include/pool.h | 1 +
src/include/protocol/pool_proto_modules.h | 5 +
src/protocol/pool_process_query.c | 282 ++++++++++++++--
src/protocol/pool_proto_modules.c | 372 +++++++--------------
src/query_cache/pool_memqcache.c | 192 ++++++++---
.../regression/tests/006.memqcache/jdbctest.java | 2 +-
src/test/regression/tests/051.bug60/test.sh | 6 +-
src/utils/pool_stream.c | 54 ++-
14 files changed, 983 insertions(+), 383 deletions(-)
diff --git a/src/Makefile.am b/src/Makefile.am
index 05338acfd..d1187bd2b 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -19,7 +19,8 @@ pgpool_SOURCES = main/main.c \
protocol/pool_process_query.c \
protocol/pool_connection_pool.c \
protocol/pool_proto_modules.c \
- query_cache/pool_memqcache.c \
+ query_cache/pool_memqcache.c \
+ protocol/CommandComplete.c \
context/pool_session_context.c \
context/pool_process_context.c \
context/pool_query_context.c \
diff --git a/src/Makefile.in b/src/Makefile.in
index 53277fc9c..9031ada27 100644
--- a/src/Makefile.in
+++ b/src/Makefile.in
@@ -111,6 +111,7 @@ am_pgpool_OBJECTS = main/main.$(OBJEXT) main/pool_globals.$(OBJEXT) \
protocol/child.$(OBJEXT) protocol/pool_process_query.$(OBJEXT) \
protocol/pool_connection_pool.$(OBJEXT) \
protocol/pool_proto_modules.$(OBJEXT) \
+ protocol/CommandComplete.$(OBJEXT) \
query_cache/pool_memqcache.$(OBJEXT) \
context/pool_session_context.$(OBJEXT) \
context/pool_process_context.$(OBJEXT) \
@@ -427,7 +428,8 @@ pgpool_SOURCES = main/main.c \
protocol/pool_process_query.c \
protocol/pool_connection_pool.c \
protocol/pool_proto_modules.c \
- query_cache/pool_memqcache.c \
+ query_cache/pool_memqcache.c \
+ protocol/CommandComplete.c \
context/pool_session_context.c \
context/pool_process_context.c \
context/pool_query_context.c \
@@ -696,6 +698,8 @@ protocol/pool_connection_pool.$(OBJEXT): protocol/$(am__dirstamp) \
protocol/$(DEPDIR)/$(am__dirstamp)
protocol/pool_proto_modules.$(OBJEXT): protocol/$(am__dirstamp) \
protocol/$(DEPDIR)/$(am__dirstamp)
+protocol/CommandComplete.$(OBJEXT): protocol/$(am__dirstamp) \
+ protocol/$(DEPDIR)/$(am__dirstamp)
query_cache/$(am__dirstamp):
@$(MKDIR_P) query_cache
@: > query_cache/$(am__dirstamp)
@@ -841,6 +845,7 @@ distclean-compile:
@AMDEP_TRUE@@am__include@ @am__quote@pcp_con/$(DEPDIR)/pcp_child.Po@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@pcp_con/$(DEPDIR)/pcp_worker.Po@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@pcp_con/$(DEPDIR)/recovery.Po@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@protocol/$(DEPDIR)/CommandComplete.Po@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@protocol/$(DEPDIR)/child.Po@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@protocol/$(DEPDIR)/pool_connection_pool.Po@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@protocol/$(DEPDIR)/pool_process_query.Po@am__quote@
diff --git a/src/context/pool_query_context.c b/src/context/pool_query_context.c
index b2ca6d615..4254a191e 100644
--- a/src/context/pool_query_context.c
+++ b/src/context/pool_query_context.c
@@ -730,7 +730,7 @@ POOL_STATUS pool_send_and_wait(POOL_QUERY_CONTEXT *query_context,
*/
POOL_STATUS pool_extended_send_and_wait(POOL_QUERY_CONTEXT *query_context,
char *kind, int len, char *contents,
- int send_type, int node_id)
+ int send_type, int node_id, bool nowait)
{
POOL_SESSION_CONTEXT *session_context;
POOL_CONNECTION *frontend;
@@ -852,6 +852,11 @@ POOL_STATUS pool_extended_send_and_wait(POOL_QUERY_CONTEXT *query_context,
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);
@@ -865,47 +870,50 @@ POOL_STATUS pool_extended_send_and_wait(POOL_QUERY_CONTEXT *query_context,
str = query_context->original_query;
}
- /* Wait for response */
- for (i=0;i<NUM_BACKENDS;i++)
+ if (!nowait)
{
- if (!VALID_BACKEND(i))
- continue;
- else if (send_type < 0 && i == node_id)
- continue;
- else if (send_type > 0 && i != node_id)
- continue;
-
- /*
- * If in master/slave mode, we do not send COMMIT/ABORT to
- * slaves/standbys if it's in I(idle) state.
- */
- if (is_commit && MASTER_SLAVE && !IS_MASTER_NODE_ID(i) && TSTATE(backend, i) == 'I')
+ /* Wait for response */
+ for (i=0;i<NUM_BACKENDS;i++)
{
- continue;
- }
+ if (!VALID_BACKEND(i))
+ continue;
+ else if (send_type < 0 && i == node_id)
+ continue;
+ else if (send_type > 0 && i != node_id)
+ continue;
- if (is_begin_read_write)
- {
- if (REAL_PRIMARY_NODE_ID == i)
- str = query_context->original_query;
- else
- str = query_context->rewritten_query;
- }
+ /*
+ * If in master/slave mode, we do not send COMMIT/ABORT to
+ * slaves/standbys if it's in I(idle) state.
+ */
+ if (is_commit && MASTER_SLAVE && !IS_MASTER_NODE_ID(i) && TSTATE(backend, i) == 'I')
+ {
+ continue;
+ }
- wait_for_query_response_with_trans_cleanup(frontend,
- CONNECTION(backend, i),
- MAJOR(backend),
- MASTER_CONNECTION(backend)->pid,
- MASTER_CONNECTION(backend)->key);
+ if (is_begin_read_write)
+ {
+ if (REAL_PRIMARY_NODE_ID == i)
+ str = query_context->original_query;
+ else
+ str = query_context->rewritten_query;
+ }
- /*
- * Check if some error detected. If so, emit
- * log. This is useful when invalid encoding error
- * occurs. In this case, PostgreSQL does not report
- * what statement caused that error and make users
- * confused.
- */
- per_node_error_log(backend, i, str, "pool_send_and_wait: Error or notice message from backend: ", true);
+ wait_for_query_response_with_trans_cleanup(frontend,
+ CONNECTION(backend, i),
+ MAJOR(backend),
+ MASTER_CONNECTION(backend)->pid,
+ MASTER_CONNECTION(backend)->key);
+
+ /*
+ * Check if some error detected. If so, emit
+ * log. This is useful when invalid encoding error
+ * occurs. In this case, PostgreSQL does not report
+ * what statement caused that error and make users
+ * confused.
+ */
+ per_node_error_log(backend, i, str, "pool_send_and_wait: Error or notice message from backend: ", true);
+ }
}
if(rewritten_begin)
diff --git a/src/context/pool_session_context.c b/src/context/pool_session_context.c
index 7b10da51f..bbc829234 100644
--- a/src/context/pool_session_context.c
+++ b/src/context/pool_session_context.c
@@ -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-2015 PgPool Global Development Group
*
* Permission to use, copy, modify, and distribute this software and
* its documentation for any purpose and without fee is hereby
@@ -35,6 +35,7 @@ static POOL_SESSION_CONTEXT session_context_d;
static POOL_SESSION_CONTEXT *session_context = NULL;
static void GetTranIsolationErrorCb(void *arg);
static void init_sent_message_list(void);
+static POOL_PENDING_MESSAGE *copy_pending_message(POOL_PENDING_MESSAGE *messag);
/*
* Initialize per session context
@@ -123,6 +124,15 @@ void pool_init_session_context(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *
session_context->query_cache_array = pool_create_query_cache_array();
session_context->num_selects = 0;
}
+
+ /* Clear sync map */
+ pool_clear_sync_map();
+
+ /* Unset pending response */
+ pool_unset_pending_response();
+
+ /* Initialize pending message list */
+ pool_pending_messages_init();
}
/*
@@ -195,6 +205,9 @@ void pool_set_query_in_progress(void)
*/
void pool_unset_query_in_progress(void)
{
+ ereport(DEBUG1,
+ (errmsg("session context: unsetting query in progress. DONE")));
+
pool_get_session_context(false)->in_progress = false;
}
@@ -304,7 +317,8 @@ bool pool_remove_sent_message(char kind, const char *name)
if (msglist->sent_messages[i]->kind == kind &&
!strcmp(msglist->sent_messages[i]->name, name))
{
- pool_sent_message_destroy(msglist->sent_messages[i]);
+ //XXX
+ //pool_sent_message_destroy(msglist->sent_messages[i]);
break;
}
}
@@ -829,3 +843,280 @@ bool can_query_context_destroy(POOL_QUERY_CONTEXT *qc)
return true;
}
+
+/*
+ * Set sync map
+ */
+void pool_set_sync_map(int node_id)
+{
+ if (!session_context)
+ ereport(ERROR,
+ (errmsg("pool_set_sync_map: session context is not initialized")));
+
+ session_context->sync_map[node_id] = true;
+}
+
+/*
+ * Check if sync map is set
+ */
+bool pool_is_set_sync_map(int node_id)
+{
+ if (!session_context)
+ ereport(ERROR,
+ (errmsg("pool_is_set_sync_map: session context is not initialized")));
+
+ return session_context->sync_map[node_id];
+}
+
+/*
+ * Get nth valid node id from sync map
+ */
+int pool_get_nth_sync_map(int nth)
+{
+ int i;
+ int cnt = 0;
+
+ for (i = 0; i < NUM_BACKENDS; i++)
+ {
+ if (!VALID_BACKEND(i))
+ continue;
+
+ if (session_context->sync_map[i])
+ {
+ if (cnt == nth)
+ return i; /* found nth valid node id */
+ }
+
+ cnt++;
+ }
+
+ return -1; /* no valid node id found */
+}
+
+/*
+ * Clear sync map
+ */
+void pool_clear_sync_map(void)
+{
+ if (!session_context)
+ ereport(ERROR,
+ (errmsg("pool_clear_sync_map: session context is not initialized")));
+
+ memset(&session_context->sync_map, 0, sizeof(session_context->sync_map));
+}
+
+/*
+ * Set pending response
+ */
+void pool_set_pending_response(void)
+{
+ if (!session_context)
+ ereport(ERROR,
+ (errmsg("pool_set_pending_response: session context is not initialized")));
+
+ session_context->is_pending_response = true;
+}
+
+/*
+ * Unset pending response
+ */
+void pool_unset_pending_response(void)
+{
+ if (!session_context)
+ ereport(ERROR,
+ (errmsg("pool_unset_pending_response: session context is not initialized")));
+
+ session_context->is_pending_response = false;
+}
+
+/*
+ * Returns true if pending response is set
+ */
+bool pool_is_pending_response(void)
+{
+ if (!STREAM)
+ return false;
+
+ if (!session_context)
+ ereport(ERROR,
+ (errmsg("pool_is_pending_response: session context is not initialized")));
+
+ return session_context->is_pending_response;
+}
+
+
+/*
+ * Initialize pending message list
+ */
+void pool_pending_messages_init(void)
+{
+ if (!session_context)
+ ereport(ERROR,
+ (errmsg("pool_pending_message_init: session context is not initialized")));
+
+ session_context->pending_messages = NIL;
+}
+
+/*
+ * Destroy pending message list
+ */
+void pool_pending_messages_destroy(void)
+{
+ ListCell *cell;
+ POOL_PENDING_MESSAGE *msg;
+
+ if (!session_context)
+ ereport(ERROR,
+ (errmsg("pool_pending_message_destory: session context is not initialized")));
+
+ foreach(cell, session_context->pending_messages)
+ {
+ msg = (POOL_PENDING_MESSAGE *) lfirst(cell);
+ pfree(msg->contents);
+ }
+ list_free(session_context->pending_messages);
+}
+
+/*
+ * Create one message.
+ */
+POOL_PENDING_MESSAGE *pool_pending_messages_create(char kind, int len, char *contents)
+{
+ POOL_PENDING_MESSAGE* msg;
+ MemoryContext old_context;
+
+ if (!session_context)
+ ereport(ERROR,
+ (errmsg("pool_pending_message_create: session context is not initialized")));
+
+ old_context = MemoryContextSwitchTo(session_context->memory_context);
+ msg = palloc(sizeof(POOL_PENDING_MESSAGE));
+
+ switch (kind)
+ {
+ case 'P':
+ msg->type = POOL_PARSE;
+ break;
+
+ case 'B':
+ msg->type = POOL_BIND;
+ break;
+
+ case 'C':
+ msg->type = POOL_CLOSE;
+ break;
+
+ default:
+ ereport(ERROR,
+ (errmsg("pool_pending_message_create: unknow kind: %c", kind)));
+ break;
+ }
+
+ msg->contents = palloc(len);
+ memcpy(msg->contents, contents, len);
+ msg->contents_len = len;
+
+ MemoryContextSwitchTo(old_context);
+
+ return msg;
+}
+
+/*
+ * Add one message
+ */
+void pool_pending_message_add(POOL_PENDING_MESSAGE* message)
+{
+ POOL_PENDING_MESSAGE* msg;
+ MemoryContext old_context;
+
+ if (!session_context)
+ ereport(ERROR,
+ (errmsg("pool_pending_message_add: session context is not initialized")));
+
+ ereport(DEBUG1,
+ (errmsg("pool_pending_message_add: message type:%d message len:%d",
+ message->type, message->contents_len)));
+
+ old_context = MemoryContextSwitchTo(session_context->memory_context);
+ msg = copy_pending_message(message);
+ session_context->pending_messages = lappend(session_context->pending_messages, msg);
+ MemoryContextSwitchTo(old_context);
+}
+
+/*
+ * Try to find the first message specified by the message type in the message
+ * list. If found, a copy of the message is returned and the message is
+ * removed the message list. If not, returns NULL.
+ */
+POOL_PENDING_MESSAGE *pool_pending_message_remove(POOL_MESSAGE_TYPE type)
+{
+ ListCell *cell;
+ ListCell *prev;
+ ListCell *next;
+ POOL_PENDING_MESSAGE *msg;
+ MemoryContext old_context;
+
+ if (!session_context)
+ ereport(ERROR,
+ (errmsg("pool_pending_message_remove: session context is not initialized")));
+
+ old_context = MemoryContextSwitchTo(session_context->memory_context);
+
+ prev = NULL;
+ msg = NULL;
+
+ for (cell = list_head(session_context->pending_messages); cell; cell = next)
+ {
+ POOL_PENDING_MESSAGE *m = (POOL_PENDING_MESSAGE *) lfirst(cell);
+
+ next = lnext(cell);
+
+ if (m->type == type)
+ {
+ msg = copy_pending_message(m);
+
+ session_context->pending_messages =
+ list_delete_cell(session_context->pending_messages, cell, prev);
+
+ break;
+ }
+ else
+ prev = cell;
+ }
+
+ MemoryContextSwitchTo(old_context);
+ return msg;
+}
+
+/*
+ * Get message specification (either statement ('S') or portal ('P')) from a
+ * close message.
+ */
+char pool_get_close_message_spec(POOL_PENDING_MESSAGE *msg)
+{
+ return *msg->contents;
+}
+
+/*
+ * Get statement or portal name from close message.
+ * The returned pointer is within "msg".
+ */
+char *pool_get_close_message_name(POOL_PENDING_MESSAGE *msg)
+{
+ return (msg->contents)+1;
+}
+
+/*
+ * Perform deep copy of POOL_PENDING_MESSAGE object in the current memory context.
+ */
+static POOL_PENDING_MESSAGE *copy_pending_message(POOL_PENDING_MESSAGE *message)
+{
+ POOL_PENDING_MESSAGE *msg;
+
+ msg = palloc(sizeof(POOL_PENDING_MESSAGE));
+ memcpy(msg, message, sizeof(POOL_PENDING_MESSAGE));
+ msg->contents = palloc(msg->contents_len);
+ memcpy(msg->contents, message->contents, msg->contents_len);
+
+ return msg;
+}
diff --git a/src/include/context/pool_query_context.h b/src/include/context/pool_query_context.h
index 56220c764..3ab68a353 100644
--- a/src/include/context/pool_query_context.h
+++ b/src/include/context/pool_query_context.h
@@ -84,7 +84,7 @@ extern void pool_setall_node_to_be_sent(POOL_QUERY_CONTEXT *query_context);
extern bool pool_multi_node_to_be_sent(POOL_QUERY_CONTEXT *query_context);
extern void pool_where_to_send(POOL_QUERY_CONTEXT *query_context, char *query, Node *node);
extern POOL_STATUS pool_send_and_wait(POOL_QUERY_CONTEXT *query_context, int send_type, int node_id);
-extern POOL_STATUS pool_extended_send_and_wait(POOL_QUERY_CONTEXT *query_context, char *kind, int len, char *contents, int send_type, int node_id);
+extern POOL_STATUS pool_extended_send_and_wait(POOL_QUERY_CONTEXT *query_context, char *kind, int len, char *contents, int send_type, int node_id, bool nowait);
extern Node *pool_get_parse_tree(void);
extern char *pool_get_query_string(void);
extern bool is_set_transaction_serializable(Node *node);
diff --git a/src/include/context/pool_session_context.h b/src/include/context/pool_session_context.h
index 1151316fb..82512150d 100644
--- a/src/include/context/pool_session_context.h
+++ b/src/include/context/pool_session_context.h
@@ -73,6 +73,29 @@ typedef struct {
POOL_SENT_MESSAGE **sent_messages;
} POOL_SENT_MESSAGE_LIST;
+/*
+ * Received message queue used in extended protocol/streaming replication
+ * mode. The queue is an FIFO, allow to de-queue in the middle of the queue
+ * however. When Parse/Bind/Close message are received, each message is
+ * en-queued. The information is used to process those response messages,
+ * when Parse complete/Bind completes and Close compete message are received
+ * because they don't have any information regarding statement/portal.
+ *
+ * The memory used for the queue lives in the session context mememory.
+ */
+
+typedef enum {
+ POOL_PARSE,
+ POOL_BIND,
+ POOL_CLOSE
+} POOL_MESSAGE_TYPE;
+
+typedef struct {
+ POOL_MESSAGE_TYPE type;
+ char *contents; /* message packet contents excluding message kind */
+ int contents_len; /* message packet length */
+} POOL_PENDING_MESSAGE;
+
/*
* Per session context:
*/
@@ -152,6 +175,32 @@ typedef struct {
*/
POOL_QUERY_CACHE_ARRAY *query_cache_array; /* pending SELECT results */
long long int num_selects; /* number of successful SELECTs in this transaction */
+
+ /*
+ * Used for managing sync message response in streaming replication
+ * mode. In streaming replication mode, there are at most two nodes
+ * involved. One is always primary, and the other (if any) could be
+ * standby. If the load balancing manager chooses the primary as the load
+ * balancing node, only the primary node is involved. Every time extended
+ * protocol is sent to backend, we set true on the member of array
+ * below. It is cleared after sync message is processed and command
+ * complete or error response message is processed.
+ */
+ bool sync_map[MAX_NUM_BACKENDS];
+
+ /*
+ * True if parse/bind/describe/close messages are sent to backend but
+ * still sync or flush is not sent. If the flag is true, do_query issues
+ * flush and stashes responses such as "parse complete" sent from backend,
+ * then inserts them to pool_read buffer after finishing the job not to
+ * confuse the message sequence.
+ */
+ bool is_pending_response;
+
+ /*
+ * Parse/Bind/Close message queue.
+ */
+ List *pending_messages;
} POOL_SESSION_CONTEXT;
extern void pool_init_session_context(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend);
@@ -193,6 +242,20 @@ extern void pool_set_command_success(void);
extern bool pool_is_command_success(void);
extern void pool_copy_prep_where(bool *src, bool *dest);
extern bool can_query_context_destroy(POOL_QUERY_CONTEXT *qc);
+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 void pool_set_pending_response(void);
+extern void pool_unset_pending_response(void);
+extern bool pool_is_pending_response(void);
+extern void pool_pending_messages_init (void);
+extern void pool_pending_messages_destroy(void);
+extern POOL_PENDING_MESSAGE *pool_pending_messages_create(char kind, int len, char *contents);
+extern void pool_pending_message_add(POOL_PENDING_MESSAGE* message);
+extern POOL_PENDING_MESSAGE *pool_pending_message_remove(POOL_MESSAGE_TYPE type);
+extern char pool_get_close_message_spec(POOL_PENDING_MESSAGE *msg);
+extern char *pool_get_close_message_name(POOL_PENDING_MESSAGE *msg);
#ifdef NOT_USED
extern void pool_add_prep_where(char *name, bool *map);
diff --git a/src/include/pool.h b/src/include/pool.h
index 6c6b92239..83c2a143b 100644
--- a/src/include/pool.h
+++ b/src/include/pool.h
@@ -340,6 +340,7 @@ extern int my_master_node_id;
#define REPLICATION (pool_config->replication_mode)
#define MASTER_SLAVE (pool_config->master_slave_mode)
#define STREAM (MASTER_SLAVE && !strcmp("stream", pool_config->master_slave_sub_mode))
+#define SLONY (MASTER_SLAVE && !strcmp("slony", pool_config->master_slave_sub_mode))
#define DUAL_MODE (REPLICATION || MASTER_SLAVE)
#define RAW_MODE (!REPLICATION && !MASTER_SLAVE)
#define MAJOR(p) MASTER_CONNECTION(p)->sp->major
diff --git a/src/include/protocol/pool_proto_modules.h b/src/include/protocol/pool_proto_modules.h
index bb245138d..ad6b089e9 100644
--- a/src/include/protocol/pool_proto_modules.h
+++ b/src/include/protocol/pool_proto_modules.h
@@ -182,4 +182,9 @@ extern void read_kind_from_one_backend(POOL_CONNECTION *frontend, POOL_CONNECTIO
extern void do_error_command(POOL_CONNECTION *backend, int major);
extern void raise_intentional_error_if_need(POOL_CONNECTION_POOL *backend);
+/*
+ * modules defined in CommandComplete.c
+ */
+extern POOL_STATUS CommandComplete(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend);
+
#endif
diff --git a/src/protocol/pool_process_query.c b/src/protocol/pool_process_query.c
index aca1cbfb4..bf4eb0b0a 100644
--- a/src/protocol/pool_process_query.c
+++ b/src/protocol/pool_process_query.c
@@ -86,7 +86,7 @@ static POOL_STATUS read_packets_and_process(POOL_CONNECTION *frontend, POOL_CONN
static bool is_all_slaves_command_complete(unsigned char *kind_list, int num_backends, int master);
/* timeout sec for pool_check_fd */
-static int timeoutsec;
+static int timeoutsec = -1;
/*
* Main module for query processing
@@ -480,7 +480,7 @@ POOL_STATUS wait_for_query_response(POOL_CONNECTION *frontend, POOL_CONNECTION *
/* Check to see if data from backend is ready */
pool_set_timeout(30);
status = pool_check_fd(backend);
- pool_set_timeout(0);
+ pool_set_timeout(-1);
if (status < 0) /* error ? */
{
@@ -556,13 +556,18 @@ POOL_STATUS send_extended_protocol_message(POOL_CONNECTION_POOL *backend,
pool_write(cp, &sendlen, sizeof(sendlen));
pool_write(cp, string, len);
- /*
- * send "Flush" message so that backend notices us
- * the completion of the command
- */
- pool_write(cp, "H", 1);
- sendlen = htonl(4);
- pool_write_and_flush(cp, &sendlen, sizeof(sendlen));
+ if (!STREAM)
+ {
+ /*
+ * send "Flush" message so that backend notices us
+ * the completion of the command
+ */
+ pool_write(cp, "H", 1);
+ sendlen = htonl(4);
+ pool_write_and_flush(cp, &sendlen, sizeof(sendlen));
+ }
+ else
+ pool_flush(cp);
return POOL_CONTINUE;
}
@@ -576,15 +581,15 @@ int synchronize(POOL_CONNECTION *cp)
}
/*
- * set timeout in seconds for pool_check_fd
- * if timeoutval < 0, we assume no timeout(wait forever).
+ * Set timeout in seconds for pool_check_fd
+ * if timeoutval < 0, we assume no timeout (wait forever).
*/
void pool_set_timeout(int timeoutval)
{
- if (timeoutval > 0)
+ if (timeoutval >= 0)
timeoutsec = timeoutval;
else
- timeoutsec = 0;
+ timeoutsec = -1;
}
/*
@@ -611,7 +616,7 @@ int pool_check_fd(POOL_CONNECTION *cp)
fd = cp->fd;
- if (timeoutsec > 0)
+ if (timeoutsec >= 0)
{
timeout.tv_sec = timeoutsec;
timeout.tv_usec = 0;
@@ -771,12 +776,19 @@ POOL_STATUS SimpleForwardToFrontend(char kind, POOL_CONNECTION *frontend,
pool_write(frontend, &sendlen, sizeof(sendlen));
pool_write_and_flush(frontend, p1, len1);
+ ereport(DEBUG1,
+ (errmsg("SimpleForwardToFrontend: packet:%c length:%d",
+ kind, len1)));
+
/* save the received result to buffer for each kind */
if (pool_config->memory_cache_enabled)
{
if (pool_is_cache_safe() && !pool_is_cache_exceeded())
{
memqcache_register(kind, frontend, p1, len1);
+ ereport(DEBUG1,
+ (errmsg("SimpleForwardToFrontend: add to memqcache buffer:%c length:%d",
+ kind, len1)));
}
}
@@ -805,6 +817,9 @@ POOL_STATUS SimpleForwardToBackend(char kind, POOL_CONNECTION *frontend,
int sendlen;
int i;
sendlen = htonl(len + 4);
+#ifdef DEBUG
+ char msgbuf[256];
+#endif
if (len == 0)
{
@@ -815,7 +830,7 @@ POOL_STATUS SimpleForwardToBackend(char kind, POOL_CONNECTION *frontend,
{
if (VALID_BACKEND(i))
{
-#ifdef NOT_USED
+#ifdef DEBUG
snprintf(msgbuf, sizeof(msgbuf), "%c message", kind);
per_node_statement_log(backend, i, msgbuf);
#endif
@@ -836,7 +851,7 @@ POOL_STATUS SimpleForwardToBackend(char kind, POOL_CONNECTION *frontend,
{
if (VALID_BACKEND(i))
{
-#ifdef NOT_USED
+#ifdef DEBUG
snprintf(msgbuf, sizeof(msgbuf), "%c message", kind);
per_node_statement_log(backend, i, msgbuf);
#endif
@@ -1023,13 +1038,18 @@ static int
pool_set_timeout(10);
+ if ((session_context = pool_get_session_context(false)))
+ {
+ pool_unset_query_in_progress();
+ }
+
if (SimpleQuery(NULL, backend, strlen(query)+1, query) != POOL_CONTINUE)
{
- pool_set_timeout(0);
+ pool_set_timeout(-1);
return -1;
}
- pool_set_timeout(0);
+ pool_set_timeout(-1);
cache = pool_get_current_cache();
if (cache)
@@ -1876,6 +1896,9 @@ void do_query(POOL_CONNECTION *backend, char *query, POOL_SELECT_RESULT **result
bool doing_extended;
int num_close_complete;
int state;
+ bool data_pushed;
+
+ data_pushed = false;
doing_extended = pool_get_session_context(true) && pool_is_doing_extended_query_message();
@@ -1907,6 +1930,56 @@ void do_query(POOL_CONNECTION *backend, char *query, POOL_SELECT_RESULT **result
static int pname_len;
int qlen;
+ /*
+ * In streaming replication mode, send flush message before going any
+ * further to retrieve and save any pending response packet from
+ * backend. The saved packets will be poped up before returning to
+ * caller. This preserves the user's expectation of packet sequence.
+ */
+ if (pool_is_pending_response())
+ {
+ pool_write(backend, "H", 1);
+ len = htonl(sizeof(len));
+ pool_write_and_flush(backend, &len, sizeof(len));
+
+ for(;;)
+ {
+ int len;
+ char *buf;
+
+ pool_set_timeout(-1);
+
+ pool_read(backend, &kind, 1);
+ pool_push(backend, &kind, 1);
+ data_pushed = true;
+
+ pool_read(backend, &len, sizeof(len));
+ pool_push(backend, &len, sizeof(len));
+
+ len = ntohl(len);
+ if ((len - sizeof(len)) > 0)
+ {
+ len -= sizeof(len);
+ buf = palloc(len);
+ pool_read(backend, buf, len);
+ pool_push(backend, buf, len);
+ }
+
+ /* check if there's any pending data */
+ if (!pool_ssl_pending(backend) && pool_read_buffer_is_empty(backend))
+ {
+ pool_set_timeout(0);
+ if (pool_check_fd(backend) != 0)
+ {
+ ereport(DEBUG1,
+ (errmsg("do_query: no pending data")));
+ pool_set_timeout(-1);
+ break;
+ }
+ }
+ }
+ }
+
if (pname_len == 0)
{
snprintf(prepared_name, sizeof(prepared_name), "pgpool%d", getpid());
@@ -2317,6 +2390,15 @@ void do_query(POOL_CONNECTION *backend, char *query, POOL_SELECT_RESULT **result
{
ereport(DEBUG1,
(errmsg("do_query: all state completed. returning")));
+
+ if (data_pushed)
+ {
+ int poplen;
+
+ pool_pop(backend, &poplen);
+ ereport(DEBUG1,
+ (errmsg("do_query: popped data len:%d", poplen)));
+ }
break;
}
}
@@ -3074,7 +3156,7 @@ void read_kind_from_backend(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *bac
* 256 is the number of distinct values expressed by unsigned char
*/
unsigned char kind;
- int trust_kind; /* decided kind */
+ int trust_kind = 0; /* decided kind */
int max_kind = 0;
double max_count = 0;
int degenerate_node_num = 0; /* number of backends degeneration requested */
@@ -3092,12 +3174,15 @@ void read_kind_from_backend(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *bac
{
read_kind_from_one_backend(frontend, backend, (char *)&kind, MASTER_NODE_ID);
+ ereport(DEBUG1,
+ (errmsg("reading backend data packet kind"),
+ errdetail("master node id: %d", MASTER_NODE_ID)));
/*
* If we received a notification message in master/slave mode,
* other backends will not receive the message.
* So we should skip other nodes otherwise we will hung in pool_read.
*/
- if (kind == 'A')
+ if (kind == 'A')
{
*decided_kind = 'A';
@@ -3128,7 +3213,23 @@ void read_kind_from_backend(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *bac
char *p, *value;
int len;
- pool_read(CONNECTION(backend, i), &kind, 1);
+ kind = 0;
+ if (pool_read(CONNECTION(backend, i), &kind, 1))
+ {
+ ereport(FATAL,
+ (return_code(2),
+ errmsg("failed to read kind from backend %d", i),
+ errdetail("pool_read retruns error")));
+ }
+
+ if (kind == 0)
+ {
+ ereport(FATAL,
+ (return_code(2),
+ errmsg("failed to read kind from backend %d", i),
+ errdetail("kind == 0")));
+ }
+
ereport(DEBUG2,
(errmsg("reading backend data packet kind"),
errdetail("backend:%d kind:'%c'",i, kind)));
@@ -3186,6 +3287,10 @@ void read_kind_from_backend(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *bac
kind_list[i] = 0;
}
+ ereport(DEBUG1,
+ (errmsg("read_kind_from_backend max_count:%f num_executed_nodes:%d",
+ max_count, num_executed_nodes)));
+
if (max_count != num_executed_nodes)
{
/*
@@ -3193,13 +3298,138 @@ void read_kind_from_backend(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *bac
*/
/*
- * However here is a special case. In master slave mode, if
- * master gets an error at commit, while other slaves are
- * normal at commit, we don't need to degenrate any backend
- * because it is likely that the error was caused by a
+ * If we are in streaming replication mode and kind = 'Z' (ready for
+ * query) on master and kind on standby is not 'Z', it is likely that
+ * following scenario happened.
+ *
+ * FE=>Parse("BEGIN")
+ * FE=>Bind
+ * FE=>Execute("BEGIN");
+ *
+ * At this point, Parse, Bind and Execute message are sent to both primary
+ * and standby node (if load balancing node is not primary). Then, the
+ * frontend sends next parse message which is supposed to be executed
+ * on primary only, for example UPDATE.
+ *
+ * FE=>Parse("UPDATE ...")
+ * FE=>Bind
+ * FE=>Execute("Update...");
+ * FE=>Sync
+ *
+ * Here, Sync message is sent to only primary, since the query context
+ * said so. Thus responses for Parse, Bind, and Execute for "BEGIN"
+ * are still penidng on the standby node.
+
+ * The the frontend sends closes the transaction.
+ *
+ * FE=>Parse("COMMIT");
+ * FE=>Bind
+ * FE=>Execute
+ * FE=>Sync
+ *
+ * Here, sync message is sent to both primary and standby. So both
+ * nodes happily send back responses. Those will be:
+ *
+ * On the primary side: parse complete, bind complete, command
+ * complete and ready for query[1]. These are all responses for COMMIT.
+ *
+ * On the standby side: parse complete, bind complete and command
+ * complete. These are all responses for *BEGIN*. Then parse
+ * complete[2], bind complete, command complete and ready for query
+ * (responses for COMMIT).
+ *
+ * As a result, [1] and [2] are different and kind mismatch error
+ * occurs.
+ *
+ * There seem to be no general solution for this at least for me. What
+ * we can do is, discard the standby response until it matches the
+ * response of the primary server. Although this could happen in
+ * various scenario, probably we should employ this strategy for 'Z'
+ * (ready for response) case only, since it's a good candidate to
+ * re-sync primary and standby.
+ */
+ if (kind_list[MASTER_NODE_ID] == 'Z' && STREAM && query_context->parse_tree &&
+ is_commit_query(query_context->parse_tree))
+ {
+ POOL_SESSION_CONTEXT *session_context;
+ POOL_CONNECTION *s;
+ char *buf;
+ int len;
+
+ session_context = pool_get_session_context(false);
+
+ s = CONNECTION(backend, session_context->load_balance_node_id);
+
+ /* skip len and contents corresponding standby data */
+ pool_read(s, &len, sizeof(len));
+ len = ntohl(len);
+ if ((len - sizeof(len)) > 0)
+ {
+ len -= sizeof(len);
+ buf = palloc(len);
+ pool_read(s, buf, len);
+ pfree(buf);
+ }
+ ereport(LOG,
+ (errmsg("read_kind_from_backend: skipped first standy packet")));
+
+ for(;;)
+ {
+ if (!pool_ssl_pending(s) && pool_read_buffer_is_empty(s))
+ {
+ pool_set_timeout(0);
+ if (pool_check_fd(s) != 0)
+ {
+ ereport(LOG,
+ (errmsg("readkind_from_backend: no pending data")));
+ pool_set_timeout(-1);
+ break;
+ }
+ }
+
+ pool_set_timeout(-1);
+
+ pool_read(s, &kind, 1);
+
+ ereport(LOG,
+ (errmsg("read_kind_from_backend: checking kind: '%c'", kind)));
+
+ if (kind == 'Z')
+ {
+ /* succeeded in re-sync */
+ ereport(LOG,
+ (errmsg("read_kind_from_backend: succeeded in re-sync")));
+ *decided_kind = kind;
+ return;
+ }
+
+ ereport(LOG,
+ (errmsg("read_kind_from_backend: reading len")));
+ pool_read(s, &len, sizeof(len));
+ ereport(LOG,
+ (errmsg("read_kind_from_backend: finished reading len:%d", ntohl(len))));
+
+ len = ntohl(len);
+ if ((len - sizeof(len)) > 0)
+ {
+ len -= sizeof(len);
+ ereport(LOG,
+ (errmsg("read_kind_from_backend: reading message len:%d", len)));
+
+ buf = palloc(len);
+ pool_read(s, buf, len);
+ pfree(buf);
+ }
+ }
+ }
+
+ /*
+ * In master slave mode, if master gets an error at commit, while
+ * other slaves are normal at commit, we don't need to degenerate any
+ * backend because it is likely that the error was caused by a
* deferred trigger.
*/
- if (MASTER_SLAVE && query_context->parse_tree &&
+ else if (MASTER_SLAVE && query_context->parse_tree &&
is_commit_query(query_context->parse_tree) &&
kind_list[MASTER_NODE_ID] == 'E' &&
is_all_slaves_command_complete(kind_list, NUM_BACKENDS, MASTER_NODE_ID))
diff --git a/src/protocol/pool_proto_modules.c b/src/protocol/pool_proto_modules.c
index 9c30a6b9a..ff04f72ab 100644
--- a/src/protocol/pool_proto_modules.c
+++ b/src/protocol/pool_proto_modules.c
@@ -87,7 +87,6 @@ static POOL_STATUS parse_before_bind(POOL_CONNECTION *frontend,
POOL_CONNECTION_POOL *backend,
POOL_SENT_MESSAGE *message);
static int* find_victim_nodes(int *ntuples, int nmembers, int master_node, int *number_of_nodes);
-static int extract_ntuples(char *message);
static POOL_STATUS close_standby_transactions(POOL_CONNECTION *frontend,
POOL_CONNECTION_POOL *backend);
/*
@@ -598,7 +597,7 @@ POOL_STATUS Execute(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
ereport(FATAL,
(return_code(2),
errmsg("unable to Execute"),
- errdetail("unable to get bind message")));
+ errdetail("unable to get query context")));
if (!bind_msg->query_context->original_query)
ereport(FATAL,
@@ -715,7 +714,7 @@ POOL_STATUS Execute(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
/* check if query is "COMMIT" or "ROLLBACK" */
commit = is_commit_or_rollback_query(node);
- if (REPLICATION)
+ if (!STREAM)
{
/*
* Query is not commit/rollback
@@ -723,7 +722,7 @@ POOL_STATUS Execute(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
if (!commit)
{
/* Send the query to master node */
- pool_extended_send_and_wait(query_context, "E", len, contents, 1, MASTER_NODE_ID);
+ pool_extended_send_and_wait(query_context, "E", len, contents, 1, MASTER_NODE_ID, false);
/* Check specific errors */
specific_error = check_errors(backend, MASTER_NODE_ID);
@@ -742,23 +741,24 @@ POOL_STATUS Execute(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
memset(msg + len, 0, sizeof(int));
/* send query to other nodes */
- pool_extended_send_and_wait(query_context, "E", len, msg, -1, MASTER_NODE_ID);
+ pool_extended_send_and_wait(query_context, "E", len, msg, -1, MASTER_NODE_ID, false);
}
else
{
- pool_extended_send_and_wait(query_context, "E", len, contents, -1, MASTER_NODE_ID);
+ pool_extended_send_and_wait(query_context, "E", len, contents, -1, MASTER_NODE_ID, false);
}
/* send "COMMIT" or "ROLLBACK" to only master node if query is "COMMIT" or "ROLLBACK" */
if (commit)
{
- pool_extended_send_and_wait(query_context, "E", len, contents, 1, MASTER_NODE_ID);
+ pool_extended_send_and_wait(query_context, "E", len, contents, 1, MASTER_NODE_ID, false);
}
}
- else
+ else /* streaming replication mode */
{
- pool_extended_send_and_wait(query_context, "E", len, contents, 1, MASTER_NODE_ID);
- pool_extended_send_and_wait(query_context, "E", len, contents, -1, MASTER_NODE_ID);
+ pool_extended_send_and_wait(query_context, "E", len, contents, 1, MASTER_NODE_ID, true);
+ pool_extended_send_and_wait(query_context, "E", len, contents, -1, MASTER_NODE_ID, true);
+ pool_unset_query_in_progress();
}
return POOL_CONTINUE;
@@ -1027,7 +1027,7 @@ POOL_STATUS Parse(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
}
}
- if (REPLICATION || MASTER_SLAVE)
+ if (REPLICATION || SLONY)
{
/*
* We must synchronize because Parse message acquires table
@@ -1035,7 +1035,7 @@ POOL_STATUS Parse(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
*/
ereport(DEBUG1,
(errmsg("Parse: waiting for master completing the query")));
- pool_extended_send_and_wait(query_context, "P", len, contents, 1, MASTER_NODE_ID);
+ pool_extended_send_and_wait(query_context, "P", len, contents, 1, MASTER_NODE_ID, false);
/*
@@ -1076,12 +1076,21 @@ POOL_STATUS Parse(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
}
else
{
- pool_extended_send_and_wait(query_context, "P", len, contents, -1, MASTER_NODE_ID);
+ pool_extended_send_and_wait(query_context, "P", len, contents, -1, MASTER_NODE_ID, false);
}
}
+ else if (STREAM)
+ {
+ /* XXX fix me:even with streaming replication mode, couldn't we have a deadlock */
+ pool_set_query_in_progress();
+ pool_extended_send_and_wait(query_context, "P", len, contents, 1, MASTER_NODE_ID, true);
+ pool_extended_send_and_wait(query_context, "P", len, contents, -1, MASTER_NODE_ID, true);
+ pool_add_sent_message(session_context->uncompleted_message);
+ pool_unset_query_in_progress();
+ }
else
{
- pool_extended_send_and_wait(query_context, "P", len, contents, 1, MASTER_NODE_ID);
+ pool_extended_send_and_wait(query_context, "P", len, contents, 1, MASTER_NODE_ID, false);
}
return POOL_CONTINUE;
@@ -1099,6 +1108,7 @@ POOL_STATUS Bind(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
POOL_SESSION_CONTEXT *session_context;
POOL_QUERY_CONTEXT *query_context;
int insert_stmt_with_lock = 0;
+ bool nowait;
/* Get session context */
session_context = pool_get_session_context(false);
@@ -1150,6 +1160,10 @@ POOL_STATUS Bind(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
session_context->query_context = query_context;
+/*
+ * Fix me
+ */
+#ifdef NOT_USED
if (pool_config->load_balance_mode && pool_is_writing_transaction())
{
pool_where_to_send(query_context, query_context->original_query,
@@ -1158,6 +1172,7 @@ POOL_STATUS Bind(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
if (parse_before_bind(frontend, backend, parse_msg) != POOL_CONTINUE)
return POOL_END;
}
+#endif
/*
* Start a transaction if necessary in replication mode
@@ -1189,12 +1204,22 @@ POOL_STATUS Bind(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
}
}
}
+
ereport(DEBUG1,
(errmsg("Bind: waiting for master completing the query")));
- pool_extended_send_and_wait(query_context, "B", len, contents, 1, MASTER_NODE_ID);
+ pool_set_query_in_progress();
+
+ nowait = (STREAM? true: false);
+
+ pool_extended_send_and_wait(query_context, "B", len, contents, 1, MASTER_NODE_ID, nowait);
+ pool_extended_send_and_wait(query_context, "B", len, contents, -1, MASTER_NODE_ID, nowait);
- pool_extended_send_and_wait(query_context, "B", len, contents, -1, MASTER_NODE_ID);
+ if (STREAM)
+ {
+ pool_add_sent_message(session_context->uncompleted_message);
+ pool_unset_query_in_progress();
+ }
if(rewrite_msg)
pfree(rewrite_msg);
@@ -1208,6 +1233,8 @@ POOL_STATUS Describe(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
POOL_SESSION_CONTEXT *session_context;
POOL_QUERY_CONTEXT *query_context;
+ bool nowait;
+
/* Get session context */
session_context = pool_get_session_context(false);
@@ -1253,9 +1280,14 @@ POOL_STATUS Describe(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
ereport(DEBUG1,
(errmsg("Describe: waiting for master completing the query")));
- pool_extended_send_and_wait(query_context, "D", len, contents, 1, MASTER_NODE_ID);
+ nowait = (REPLICATION? false: true);
- pool_extended_send_and_wait(query_context, "D", len, contents, -1, MASTER_NODE_ID);
+ pool_set_query_in_progress();
+ pool_extended_send_and_wait(query_context, "D", len, contents, 1, MASTER_NODE_ID, nowait);
+ pool_extended_send_and_wait(query_context, "D", len, contents, -1, MASTER_NODE_ID, nowait);
+
+ if (!REPLICATION)
+ pool_unset_query_in_progress();
return POOL_CONTINUE;
}
@@ -1288,9 +1320,10 @@ POOL_STATUS Close(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
(return_code(2),
errmsg("unable to execute close, invalid message")));
/*
- * As per the postgresql, calling close on non existing portals is not
- * an error. So on the same footings we will ignore all such calls and
- * return the close complete message to clients with out going to backend
+ * As per the postgresql, calling close on non existing portals or
+ * statements is not an error. So on the same footings we will ignore all
+ * such calls and return the close complete message to clients with out
+ * going to backend
*/
if (!msg)
{
@@ -1318,9 +1351,25 @@ POOL_STATUS Close(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend,
ereport(DEBUG1,
(errmsg("Close: waiting for master completing the query")));
- pool_extended_send_and_wait(query_context, "C", len, contents, 1, MASTER_NODE_ID);
- pool_extended_send_and_wait(query_context, "C", len, contents, -1, MASTER_NODE_ID);
+ pool_set_query_in_progress();
+
+ if (REPLICATION)
+ {
+ pool_extended_send_and_wait(query_context, "C", len, contents, 1, MASTER_NODE_ID, false);
+ pool_extended_send_and_wait(query_context, "C", len, contents, -1, MASTER_NODE_ID, false);
+ }
+ else
+ {
+ POOL_PENDING_MESSAGE *pmsg;
+
+ pool_extended_send_and_wait(query_context, "C", len, contents, 1, MASTER_NODE_ID, true);
+ pool_extended_send_and_wait(query_context, "C", len, contents, -1, MASTER_NODE_ID, true);
+
+ pmsg = pool_pending_messages_create('C', len, contents);
+ pool_pending_message_add(pmsg);
+ pool_unset_query_in_progress();
+ }
return POOL_CONTINUE;
}
@@ -1776,6 +1825,8 @@ POOL_STATUS CloseComplete(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backe
{
POOL_SESSION_CONTEXT *session_context;
POOL_STATUS status;
+ char kind = ' ';
+ char *name = "";
/* Get session context */
session_context = pool_get_session_context(false);
@@ -1784,223 +1835,42 @@ POOL_STATUS CloseComplete(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backe
status = SimpleForwardToFrontend('3', frontend, backend);
/* Remove the target message */
- if (session_context->uncompleted_message)
- {
- pool_remove_sent_message(session_context->uncompleted_message->kind,
- session_context->uncompleted_message->name);
- session_context->uncompleted_message = NULL;
- }
- else
- {
- ereport(ERROR,
- (errmsg("processing CloseComplete, uncompleted message not found")));
- }
-
- return status;
-}
-
-POOL_STATUS CommandComplete(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend)
-{
- int i;
- int len, len1, sendlen;
- int rows;
- char *p, *p1;
- POOL_SESSION_CONTEXT *session_context;
-
- /* Get session context */
- session_context = pool_get_session_context(false);
-
- if (session_context->query_context != NULL)
+ if (STREAM)
{
- Node *node = session_context->query_context->parse_tree;
+ POOL_PENDING_MESSAGE *pmsg;
- if (IsA(node, PrepareStmt))
- {
- if (session_context->uncompleted_message)
- {
- pool_add_sent_message(session_context->uncompleted_message);
- session_context->uncompleted_message = NULL;
- }
- }
- else if (IsA(node, DeallocateStmt))
- {
- char *name;
-
- name = ((DeallocateStmt *)node)->name;
- if (name == NULL)
- {
- pool_remove_sent_messages('Q');
- pool_remove_sent_messages('P');
- }
- else
- {
- pool_remove_sent_message('Q', name);
- pool_remove_sent_message('P', name);
- }
- }
- else if (IsA(node, DiscardStmt))
- {
- DiscardStmt *stmt = (DiscardStmt *)node;
+ pmsg = pool_pending_message_remove(POOL_CLOSE);
- if (stmt->target == DISCARD_PLANS)
- {
- pool_remove_sent_messages('Q');
- pool_remove_sent_messages('P');
- }
- else if (stmt->target == DISCARD_ALL)
- {
- pool_clear_sent_message_list();
- }
- }
- /*
- * JDBC driver sends "BEGIN" query internally if
- * setAutoCommit(false). But it does not send Sync message
- * after "BEGIN" query. In extended query protocol,
- * PostgreSQL returns ReadyForQuery when a client sends Sync
- * message. Problem is, pgpool can't know the transaction
- * state without receiving ReadyForQuery. So we remember that
- * we need to send Sync message internally afterward, whenever
- * we receive BEGIN in extended protocol.
- */
- else if (IsA(node, TransactionStmt))
+ if (pmsg)
{
- TransactionStmt *stmt = (TransactionStmt *) node;
-
- if (stmt->kind == TRANS_STMT_BEGIN || stmt->kind == TRANS_STMT_START)
- {
- int i;
-
- for (i = 0; i < NUM_BACKENDS; i++)
- {
- if (!VALID_BACKEND(i))
- continue;
-
- TSTATE(backend, i) = 'T';
- }
- pool_unset_writing_transaction();
- pool_unset_failed_transaction();
- pool_unset_transaction_isolation();
- }
+ kind = pool_get_close_message_spec(pmsg);
+ name = pool_get_close_message_name(pmsg);
}
-
}
-
- pool_read(MASTER(backend), &len, sizeof(len));
-
- len = ntohl(len);
- len -= 4;
- len1 = len;
-
- p = pool_read2(MASTER(backend), len);
- if (p == NULL)
- return POOL_END;
- p1 = palloc(len);
- memcpy(p1, p, len);
-
- rows = extract_ntuples(p);
-
- /*
- * Save number of affected tuples of master node.
- */
- session_context->ntuples[MASTER_NODE_ID] = rows;
-
- for (i=0;i<NUM_BACKENDS;i++)
+ else
{
- if (!IS_MASTER_NODE_ID(i))
+ if (session_context->uncompleted_message)
{
- if (!VALID_BACKEND(i))
- {
- session_context->ntuples[i] = -1;
- continue;
- }
-
- pool_read(CONNECTION(backend, i), &len, sizeof(len));
-
- len = ntohl(len);
- len -= 4;
-
- p = pool_read2(CONNECTION(backend, i), len);
- if (p == NULL)
- return POOL_END;
-
- if (len != len1)
- {
- ereport(DEBUG1,
- (errmsg("processing command complete"),
- errdetail("length does not match between backends master(%d) %d th backend(%d)",
- len, i, len1)));
- }
-
- int n = extract_ntuples(p);
-
- /*
- * Save number of affected tuples.
- */
- session_context->ntuples[i] = n;
-
- if (rows != n)
- {
- /*
- * Remember that we have different number of UPDATE/DELETE
- * affected tuples in backends.
- */
- session_context->mismatch_ntuples = true;
- }
+ kind = session_context->uncompleted_message->kind;
+ name = session_context->uncompleted_message->name;
+ session_context->uncompleted_message = NULL;
}
- }
-
- if (session_context->mismatch_ntuples)
- {
- char msgbuf[128];
-
- String *msg = init_string("pgpool detected difference of the number of inserted, updated or deleted tuples. Possible last query was: \"");
- string_append_char(msg, query_string_buffer);
- string_append_char(msg, "\"");
- pool_send_error_message(frontend, MAJOR(backend),
- "XX001", msg->data, "",
- "check data consistency between master and other db node", __FILE__, __LINE__);
- ereport(LOG,
- (errmsg("%s", msg->data)));
- free_string(msg);
-
- msg = init_string("CommandComplete: Number of affected tuples are:");
-
- for (i=0;i<NUM_BACKENDS;i++)
- {
- snprintf(msgbuf, sizeof(msgbuf), " %d", session_context->ntuples[i]);
- string_append_char(msg, msgbuf);
- }
- ereport(LOG,
- (errmsg("processing command complete"),
- errdetail("%s", msg->data)));
-
- free_string(msg);
- }
- else
- {
- pool_write(frontend, "C", 1);
- sendlen = htonl(len1+4);
- pool_write(frontend, &sendlen, sizeof(sendlen));
- pool_write_and_flush(frontend, p1, len1);
- }
-
- /* Save the received result to buffer for each kind */
- if (pool_config->memory_cache_enabled)
- {
- if (pool_is_cache_safe() && !pool_is_cache_exceeded())
+ else
{
- memqcache_register('C', frontend, p1, len1);
+ ereport(ERROR,
+ (errmsg("processing CloseComplete, uncompleted message not found")));
}
}
-
- pfree(p1);
-
- if (pool_is_doing_extended_query_message())
+
+ if (kind == ' ')
{
- pool_set_query_state(session_context->query_context, POOL_EXECUTE_COMPLETE);
+ pool_remove_sent_message(kind, name);
+ ereport(DEBUG1,
+ (errmsg("CloseComplete: remove uncompleted message. kind:%c, name:%s",
+ kind, name)));
}
- return POOL_CONTINUE;
+ return status;
}
POOL_STATUS ParameterDescription(POOL_CONNECTION *frontend,
@@ -2354,29 +2224,39 @@ POOL_STATUS ProcessFrontendResponse(POOL_CONNECTION *frontend,
case 'P': /* Parse */
allow_close_transaction = 0;
pool_set_doing_extended_query_message();
+#ifdef NOT_USED
if (!pool_is_query_in_progress() && !pool_is_ignore_till_sync())
pool_set_query_in_progress();
+#endif
status = Parse(frontend, backend, len, contents);
+ pool_set_pending_response();
break;
case 'B': /* Bind */
pool_set_doing_extended_query_message();
+#ifdef NOT_USED
if (!pool_is_query_in_progress() && !pool_is_ignore_till_sync())
pool_set_query_in_progress();
+#endif
status = Bind(frontend, backend, len, contents);
+ pool_set_pending_response();
break;
case 'C': /* Close */
if (!pool_is_query_in_progress() && !pool_is_ignore_till_sync())
pool_set_query_in_progress();
status = Close(frontend, backend, len, contents);
+ pool_set_pending_response();
break;
case 'D': /* Describe */
pool_set_doing_extended_query_message();
+#ifdef NOT_USED
if (!pool_is_query_in_progress() && !pool_is_ignore_till_sync())
pool_set_query_in_progress();
+#endif
status = Describe(frontend, backend, len, contents);
+ pool_set_pending_response();
break;
case 'S': /* Sync */
@@ -2386,6 +2266,7 @@ POOL_STATUS ProcessFrontendResponse(POOL_CONNECTION *frontend,
if (!pool_is_query_in_progress())
pool_set_query_in_progress();
status = SimpleForwardToBackend(fkind, frontend, backend, len, contents);
+ pool_unset_pending_response();
break;
case 'F': /* FunctionCall */
@@ -2501,19 +2382,22 @@ POOL_STATUS ProcessBackendResponse(POOL_CONNECTION *frontend,
case '1': /* ParseComplete */
status = ParseComplete(frontend, backend);
pool_set_command_success();
- pool_unset_query_in_progress();
+ if (REPLICATION||RAW_MODE)
+ pool_unset_query_in_progress();
break;
case '2': /* BindComplete */
status = BindComplete(frontend, backend);
pool_set_command_success();
- pool_unset_query_in_progress();
+ if (REPLICATION||RAW_MODE)
+ pool_unset_query_in_progress();
break;
case '3': /* CloseComplete */
status = CloseComplete(frontend, backend);
pool_set_command_success();
- pool_unset_query_in_progress();
+ if (REPLICATION||RAW_MODE)
+ pool_unset_query_in_progress();
break;
case 'E': /* ErrorResponse */
@@ -2532,7 +2416,7 @@ POOL_STATUS ProcessBackendResponse(POOL_CONNECTION *frontend,
case 'C': /* CommandComplete */
status = CommandComplete(frontend, backend);
pool_set_command_success();
- if (pool_is_doing_extended_query_message())
+ if ((REPLICATION || RAW_MODE) && pool_is_doing_extended_query_message())
pool_unset_query_in_progress();
break;
@@ -2542,25 +2426,25 @@ POOL_STATUS ProcessBackendResponse(POOL_CONNECTION *frontend,
case 'I': /* EmptyQueryResponse */
status = SimpleForwardToFrontend(kind, frontend, backend);
- if (pool_is_doing_extended_query_message())
+ if ((REPLICATION || RAW_MODE) && pool_is_doing_extended_query_message())
pool_unset_query_in_progress();
break;
case 'T': /* RowDescription */
status = SimpleForwardToFrontend(kind, frontend, backend);
- if (pool_is_doing_extended_query_message())
+ if ((REPLICATION || RAW_MODE) && pool_is_doing_extended_query_message())
pool_unset_query_in_progress();
break;
case 'n': /* NoData */
status = SimpleForwardToFrontend(kind, frontend, backend);
- if (pool_is_doing_extended_query_message())
+ if ((REPLICATION || RAW_MODE) && pool_is_doing_extended_query_message())
pool_unset_query_in_progress();
break;
case 's': /* PortalSuspended */
status = SimpleForwardToFrontend(kind, frontend, backend);
- if (pool_is_doing_extended_query_message())
+ if ((REPLICATION || RAW_MODE) && pool_is_doing_extended_query_message())
pool_unset_query_in_progress();
break;
@@ -3051,7 +2935,7 @@ static POOL_STATUS parse_before_bind(POOL_CONNECTION *frontend,
(errmsg("parse before bind"),
errdetail("waiting for backend %d completing parse", i)));
- pool_extended_send_and_wait(qc, "P", len, contents, 1, i);
+ pool_extended_send_and_wait(qc, "P", len, contents, 1, i, false);
}
else
{
@@ -3203,23 +3087,3 @@ static int* find_victim_nodes(int *ntuples, int nmembers, int master_node, int *
return victim_nodes;
}
-
-/*
- * Extract the number of tuples from CommandComplete message
- */
-static int extract_ntuples(char *message)
-{
- char *rows;
-
- if ((rows = strstr(message, "UPDATE")) || (rows = strstr(message, "DELETE")))
- rows +=7;
- else if ((rows = strstr(message, "INSERT")))
- {
- rows += 7;
- while (*rows && *rows != ' ') rows++;
- }
- else
- return 0;
-
- return atoi(rows);
-}
diff --git a/src/query_cache/pool_memqcache.c b/src/query_cache/pool_memqcache.c
index a3128949a..284b0dbaa 100644
--- a/src/query_cache/pool_memqcache.c
+++ b/src/query_cache/pool_memqcache.c
@@ -121,6 +121,7 @@ static volatile POOL_HASH_ELEMENT *get_new_hash_element(void);
static void put_back_hash_element(volatile POOL_HASH_ELEMENT *element);
static bool is_free_hash_element(void);
static char *get_relation_without_alias(RangeVar *relation);
+static void forward_pending_data(POOL_CONNECTION *frontend,POOL_CONNECTION *backend);
/*
* Connect to Memcached
@@ -614,77 +615,93 @@ POOL_STATUS pool_fetch_from_memory_cache(POOL_CONNECTION *frontend,
pool_shmem_unlock();
POOL_SETMASK(&oldmask);
- if (sts == 0)
+ if (sts != 0)
+ /* Cache not found */
+ return POOL_CONTINUE;
+
+ /*
+ * Cache found. If we are doing extended query and in streaming
+ * replication mode, we need to retrieve any responses from backend and
+ * forward them to frontend.
+ */
+ if (pool_is_doing_extended_query_message() && STREAM)
{
- /*
- * Cache found. send each messages to frontend
- */
- send_cached_messages(frontend, qcache, qcachelen);
- pfree(qcache);
+ POOL_SESSION_CONTEXT *session_context;
+ POOL_CONNECTION *target_backend;
- /*
- * If we are doing extended query, wait and discard Sync
- * message from frontend. This is necessary to prevent
- * receiving Sync message after Sending Ready for query.
- */
- if (pool_is_doing_extended_query_message())
- {
- char kind;
- int32 len;
+ ereport(DEBUG1,
+ (errmsg("memcache: fetching forwarding_pending_data")));
- if (pool_flush(frontend))
- return POOL_END;
- if (pool_read(frontend, &kind, 1))
- return POOL_END;
+ session_context = pool_get_session_context(true);
+ target_backend = CONNECTION(backend, session_context->load_balance_node_id);
+ forward_pending_data(frontend, target_backend);
+ }
- ereport(DEBUG2,
- (errmsg("memcache: fetching from memory cache: expecting sync: kind '%c'", kind)));
- if (pool_read(frontend, &len, sizeof(len)))
- return POOL_END;
- }
+ /*
+ * Send each messages to frontend
+ */
+ send_cached_messages(frontend, qcache, qcachelen);
+ pfree(qcache);
- /*
- * send a "READY FOR QUERY"
- */
- if (MAJOR(backend) == PROTO_MAJOR_V3)
- {
- signed char state;
+ /*
+ * If we are doing extended query, wait and discard Sync
+ * message from frontend. This is necessary to prevent
+ * receiving Sync message after Sending Ready for query.
+ */
+ if (pool_is_doing_extended_query_message())
+ {
+ char kind;
+ int32 len;
- /*
- * We keep previous transaction state.
- */
- state = MASTER(backend)->tstate;
- send_message(frontend, 'Z', 5, (char *)&state);
- }
- else
- {
- pool_write(frontend, "Z", 1);
- }
if (pool_flush(frontend))
- {
return POOL_END;
- }
+ if (pool_read(frontend, &kind, 1))
+ return POOL_END;
- *foundp = true;
+ ereport(DEBUG2,
+ (errmsg("memcache: fetching from memory cache: expecting sync: kind '%c'", kind)));
+ if (pool_read(frontend, &len, sizeof(len)))
+ return POOL_END;
+ }
- if (pool_config->log_per_node_statement)
- ereport(LOG,
+ /*
+ * send a "READY FOR QUERY"
+ */
+ if (MAJOR(backend) == PROTO_MAJOR_V3)
+ {
+ signed char state;
+
+ /*
+ * We keep previous transaction state.
+ */
+ state = MASTER(backend)->tstate;
+ send_message(frontend, 'Z', 5, (char *)&state);
+ }
+ else
+ {
+ pool_write(frontend, "Z", 1);
+ }
+ if (pool_flush(frontend))
+ {
+ return POOL_END;
+ }
+
+ *foundp = true;
+
+ if (pool_config->log_per_node_statement)
+ ereport(LOG,
(errmsg("fetch from memory cache"),
- errdetail("query result fetched from cache. statement: %s", contents)));
+ errdetail("query result fetched from cache. statement: %s", contents)));
- ereport(DEBUG1,
+ ereport(DEBUG1,
(errmsg("fetch from memory cache"),
- errdetail("query result found in the query cache, %s", contents)));
+ errdetail("query result found in the query cache, %s", contents)));
- return POOL_CONTINUE;
- }
-
- /* Cache not found */
return POOL_CONTINUE;
}
/*
- * Simple and rough(thus unreliable) check if the query is likely
+ * Simple and rough (thus unreliable) check if the query is likely
* SELECT. Just check if the query starts with SELECT or WITH. This
* can be used before parse tree is available.
*/
@@ -3926,3 +3943,72 @@ POOL_SHMEM_STATS *pool_get_shmem_storage_stats(void)
return &mystats;
}
+
+/*
+ * Send flash message to backend and forward any response to frontend.
+ */
+void forward_pending_data(POOL_CONNECTION *frontend, POOL_CONNECTION *backend)
+{
+ char kind;
+ int len;
+ char *buf;
+
+ /* If are doing extended query and we are in streaming replication
+ * mode, we need to retrieve any response from backend.
+ */
+ if (!pool_is_pending_response())
+ {
+ ereport(DEBUG1,
+ (errmsg("forward_pending_data: no peding data")));
+ }
+
+ /* Send flush messsage to backend to retrieve response of backend */
+ pool_write(backend, "H", 1);
+ len = htonl(sizeof(len));
+ pool_write_and_flush(backend, &len, sizeof(len));
+
+ /*
+ * Then forward any response from backend
+ */
+ for(;;)
+ {
+ /* check if there's any pending data */
+ if (!pool_ssl_pending(backend) && pool_read_buffer_is_empty(backend))
+ {
+ pool_set_timeout(0);
+ if (pool_check_fd(backend) != 0)
+ {
+ ereport(DEBUG1,
+ (errmsg("forward_pending_data: no pending data")));
+ pool_set_timeout(-1);
+ break;
+ }
+ }
+
+ pool_set_timeout(-1);
+
+ pool_read(backend, &kind, 1);
+ ereport(DEBUG1,
+ (errmsg("forward_pending_data: forwarding kind: '%c'", kind)));
+ pool_write(frontend, &kind, 1);
+
+ ereport(DEBUG1,
+ (errmsg("forward_pending_data: reading len")));
+ pool_read(backend, &len, sizeof(len));
+ ereport(DEBUG1,
+ (errmsg("forward_pending_data: finished reading len:%d", ntohl(len))));
+ pool_write(frontend, &kind, 1);
+
+ len = ntohl(len);
+ if ((len - sizeof(len)) > 0)
+ {
+ len -= sizeof(len);
+ ereport(DEBUG1,
+ (errmsg("forward_pending_data: saving message len:%d", len)));
+
+ buf = palloc(len);
+ pool_read(backend, buf, len);
+ pool_write(frontend, &kind, 1);
+ }
+ }
+}
diff --git a/src/test/regression/tests/006.memqcache/jdbctest.java b/src/test/regression/tests/006.memqcache/jdbctest.java
index 29ecb9897..479e97f64 100644
--- a/src/test/regression/tests/006.memqcache/jdbctest.java
+++ b/src/test/regression/tests/006.memqcache/jdbctest.java
@@ -26,7 +26,7 @@ public class jdbctest {
prest.close();
/*
- * Cache test in a explicit transaction
+ * Cache test in an explicit transaction
*/
conn.setAutoCommit(false);
// execute DML. This should prevent SELECTs from using query cache in the transaction.
diff --git a/src/test/regression/tests/051.bug60/test.sh b/src/test/regression/tests/051.bug60/test.sh
index b09d96d71..4032153b9 100755
--- a/src/test/regression/tests/051.bug60/test.sh
+++ b/src/test/regression/tests/051.bug60/test.sh
@@ -26,12 +26,12 @@ cd $TESTDIR
# create test environment
echo -n "creating test environment..."
-$PGPOOL_SETUP -m s -n 2 --no-stop || exit 1
+$PGPOOL_SETUP -m s -n 2 || exit 1
echo "done."
source ./bashrc.ports
-#./startall
+./startall
wait_for_pgpool_startup
export PGPORT=$PGPOOL_PORT
@@ -73,7 +73,7 @@ EOF
#
grep "kind mismatch among backends." log/pgpool.log >/dev/null 2>&1
if [ $? = 0 ];then
- # cache found! The bug reveals.
+ # kind mismatch message found! The bug reveals.
exit 1
fi
exit 0
diff --git a/src/utils/pool_stream.c b/src/utils/pool_stream.c
index f84d355e5..f5e0376c4 100644
--- a/src/utils/pool_stream.c
+++ b/src/utils/pool_stream.c
@@ -50,6 +50,9 @@ static int mystrlinelen(char *str, int upper, int *flag);
static int save_pending_data(POOL_CONNECTION *cp, void *data, int len);
static int consume_pending_data(POOL_CONNECTION *cp, void *data, int len);
static MemoryContext SwitchToConnectionContext(bool backend_connection);
+#ifdef DEBUG
+static void dump_buffer(char *buf, int len);
+#endif
static MemoryContext
SwitchToConnectionContext(bool backend_connection)
@@ -173,10 +176,22 @@ int pool_read(POOL_CONNECTION *cp, void *buf, int len)
}
}
- if (cp->ssl_active > 0) {
+ if (cp->ssl_active > 0)
+ {
readlen = pool_ssl_read(cp, readbuf, READBUFSZ);
- } else {
+ }
+ else
+ {
readlen = read(cp->fd, readbuf, READBUFSZ);
+ if (cp->isbackend)
+ {
+ ereport(DEBUG1,
+ (errmsg("pool_read: read %d bytes from backend %d",
+ readlen, cp->db_node_id)));
+#ifdef DEBUG
+ dump_buffer(readbuf, readlen);
+#endif
+ }
}
if (readlen == -1)
@@ -310,10 +325,17 @@ char *pool_read2(POOL_CONNECTION *cp, int len)
}
}
- if (cp->ssl_active > 0) {
+ if (cp->ssl_active > 0)
+ {
readlen = pool_ssl_read(cp, buf, len);
- } else {
+ }
+ else
+ {
readlen = read(cp->fd, buf, len);
+ if (cp->isbackend)
+ ereport(DEBUG1,
+ (errmsg("pool_read2: read %d bytes from backend %d",
+ readlen, cp->db_node_id)));
}
if (readlen == -1)
@@ -398,6 +420,16 @@ int pool_write_noerror(POOL_CONNECTION *cp, void *buf, int len)
if (cp->no_forward)
return 0;
+
+ if (len == 1 && cp->isbackend)
+ {
+ char c;
+
+ c = ((char *)buf)[0];
+
+ ereport(DEBUG1,
+ (errmsg("pool_write: to backend: kind:%c", c)));
+ }
while (len > 0)
{
@@ -1099,3 +1131,17 @@ void pool_unset_nonblock(int fd)
errdetail("fcntl system call failed with error \"%s\"", strerror(errno))));
}
}
+
+#ifdef DEBUG
+/*
+ * Debug aid
+ */
+static void dump_buffer(char *buf, int len)
+{
+ while (--len)
+ {
+ ereport(DEBUG1,
+ (errmsg("%02x", *buf++)));
+ }
+}
+#endif
[parent: 426dd30114a7]