pgpool2.git / summary / log / commit / refs
commit dc5b7ccc163af5319d02558279a4e145812e0f7d
Author: Tatsuo Ishii <ishii@postgresql.org>
Date: Wed Jan 16 04:22:28 2013 +0000
Fix LISTEN/NOTIFY handling bugs reported in bug #45.
1) In streaming replication mode:
Session 1: LISTEN aaa;
Session 2: NOTIFY aaa;
Session 1: LISTEN aaa; --- hangs
(If LISTEN and NOTIFY are issued in a same session, it works fine.)
We assume that packets come from all nodes. However in streaming
replication mode, notification message only comes from primary
node and we should avoid reading from standby nodes.
2) In streaming replication mode: If primary node is not node 0, it
hangs like #1 even if fix applies. This is because MASTER_NODE_ID
macro (actually pool_virtual_master_db_node_id() always returns
REAL_MASTER_NODE_ID, which is node 0 (if it is alive). The function
should return PRIMARY_NODE_ID in master/slave mode.
3) In replication mode, LISTEN/NOTIFY simply does not work. In the
mode, NOTIFY is sent to all backends. However the order of arrival of
'Notification response' is not necessarily the master first and then
slaves. So if it arrives slave first, we should try to read from
master, rather than just discard it. Fixed in pool_process_query().
4) In replication mode, if LISTEN and NOTIFY are issued in a same
session, the session is disconnected because do_command() may receive
other than 'N', 'E', 'S' and 'C'. The solution is, put 'A' packet on a
stack and pop out when it is convenient. For this purpose, new
functions pool_push(), pool_pop() and pool_stacklen() are added.
Considering the size of the patch, I do not back patch to 3.1 or
before(so far, we have not heard any complaints on 3.1 or before).
pool.h | 3 +
pool_process_query.c | 155 ++++++++++++++++++++++++++++++++++++++++-----------
pool_query_context.c | 8 ++-
pool_stream.c | 60 ++++++++++++++++++++
pool_stream.h | 5 +-
5 files changed, 195 insertions(+), 36 deletions(-)
diff --git a/pool.h b/pool.h
index d4830d618..56458ee69 100644
--- a/pool.h
+++ b/pool.h
@@ -164,6 +164,9 @@ typedef struct {
char *buf2; /* buffer for pool_read2 */
int bufsz2; /* its size in bytes */
+ char *buf3; /* buffer for pool_push/pop */
+ int bufsz3; /* its size in bytes */
+
int isbackend; /* this connection is for backend if non 0 */
int db_node_id; /* DB node id for this connection */
diff --git a/pool_process_query.c b/pool_process_query.c
index 69e8ecab9..1dfce5f69 100644
--- a/pool_process_query.c
+++ b/pool_process_query.c
@@ -175,6 +175,20 @@ POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
return POOL_END;
}
+ /*
+ * If we are not processing a query, now is the time to
+ * extract retrieve pending data from buffer stack if any.
+ */
+ if (!pool_is_query_in_progress())
+ {
+ for (i=0;i<NUM_BACKENDS;i++)
+ {
+ int plen;
+ if (VALID_BACKEND(i) && pool_stacklen(CONNECTION(backend, i)) > 0)
+ pool_pop(CONNECTION(backend, i), &plen);
+ }
+ }
+
/*
* If we are prcessing query, process it.
*/
@@ -291,30 +305,52 @@ POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
pool_error("pool_process_query: error while reading message kind from backend %d", i);
return POOL_ERROR;
}
- pool_log("pool_process_query: discard %c packet from backend %d", kind, i);
- if (MAJOR(backend) == PROTO_MAJOR_V3)
+ if (kind == 'A')
{
- if (pool_read(CONNECTION(backend, i), &len, sizeof(len)) < 0)
+ /*
+ * In replication mode, NOTIFY is sent to all backends.
+ * However the order of arrival of 'Notification response'
+ * is not necessarily the master first and then slaves.
+ * So if it arrives slave first, we should try to read from master,
+ * rather than just discard it.
+ */
+ pool_unread(CONNECTION(backend, i), &kind, sizeof(kind));
+ pool_log("pool_process_query: received %c packet from backend %d. Don't dicard and read %c packet from master", kind, i, kind);
+ if (pool_read(CONNECTION(backend, MASTER_NODE_ID), &kind, sizeof(kind)) < 0)
{
- pool_error("pool_process_query: error while reading message length from backend %d", i);
- return POOL_ERROR;
- }
- 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);
+ pool_error("pool_process_query: error while reading message kind from backend %d", MASTER_NODE_ID);
return POOL_ERROR;
}
+ pool_unread(CONNECTION(backend, MASTER_NODE_ID), &kind, sizeof(kind));
}
else
{
- string = pool_read_string(CONNECTION(backend, i), &len, 0);
- 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_ERROR;
+ if (pool_read(CONNECTION(backend, i), &len, sizeof(len)) < 0)
+ {
+ pool_error("pool_process_query: error while reading message length from backend %d", i);
+ return POOL_ERROR;
+ }
+ 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_ERROR;
+ }
+ }
+ else
+ {
+ 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_ERROR;
+ }
}
}
}
@@ -1075,28 +1111,36 @@ POOL_STATUS SimpleForwardToFrontend(char kind, POOL_CONNECTION *frontend,
}
memcpy(p1, p, len);
- for (i=0;i<NUM_BACKENDS;i++)
+ /*
+ * 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 (!MASTER_SLAVE || kind != 'A')
{
- if (VALID_BACKEND(i) && !IS_MASTER_NODE_ID(i))
+ for (i=0;i<NUM_BACKENDS;i++)
{
- status = pool_read(CONNECTION(backend, i), &len, sizeof(len));
- if (status < 0)
+ if (VALID_BACKEND(i) && !IS_MASTER_NODE_ID(i))
{
- pool_error("SimpleForwardToFrontend: error while reading message length");
- return POOL_END;
- }
+ status = pool_read(CONNECTION(backend, i), &len, sizeof(len));
+ if (status < 0)
+ {
+ pool_error("SimpleForwardToFrontend: error while reading message length");
+ return POOL_END;
+ }
- len = ntohl(len);
- len -= 4;
+ len = ntohl(len);
+ len -= 4;
- p = pool_read2(CONNECTION(backend, i), len);
- if (p == NULL)
- return POOL_END;
+ p = pool_read2(CONNECTION(backend, i), len);
+ if (p == NULL)
+ return POOL_END;
- if (len != len1)
- {
- pool_debug("SimpleForwardToFrontend: length does not match between backends master(%d) %d th backend(%d) kind:(%c)",
- len, i, len1, kind);
+ if (len != len1)
+ {
+ pool_debug("SimpleForwardToFrontend: length does not match between backends master(%d) %d th backend(%d) kind:(%c)",
+ len, i, len1, kind);
+ }
}
}
}
@@ -1823,11 +1867,12 @@ POOL_STATUS do_command(POOL_CONNECTION *frontend, POOL_CONNECTION *backend,
pool_error("do_command: error while reading message length");
return POOL_END;
}
+ pool_debug("len:%x", len);
len = ntohl(len) - 4;
-
- if (kind != 'N' && kind != 'E' && kind != 'S' && kind != 'C')
+
+ if (kind != 'N' && kind != 'E' && kind != 'S' && kind != 'C' && kind != 'A')
{
- pool_error("do_command: error, kind is not N, E, S or C(%02x)", kind);
+ pool_error("do_command: error, kind is not N, E, S, C or A(%02x)", kind);
return POOL_END;
}
string = pool_read2(backend, len);
@@ -1836,6 +1881,22 @@ POOL_STATUS do_command(POOL_CONNECTION *frontend, POOL_CONNECTION *backend,
pool_error("do_command: error while reading rest of message");
return POOL_END;
}
+
+ /*
+ * It is possible that we receives a notification response
+ * ('A') from one of backends prior to "ready for query"
+ * response if LISTEN and NOTIFY are issued in a same
+ * connection. So we need to save notification response to
+ * stack buffer so that we could retrieve it later on.
+ */
+ if (kind == 'A')
+ {
+ int nlen = htonl(len+4);
+ pool_debug("nlen:%x", nlen);
+ pool_push(backend, &kind, sizeof(kind));
+ pool_push(backend, &nlen, sizeof(nlen));
+ pool_push(backend, string, len);
+ }
}
else
{
@@ -3561,6 +3622,7 @@ POOL_STATUS read_kind_from_backend(POOL_CONNECTION *frontend, POOL_CONNECTION_PO
double max_count = 0;
int degenerate_node_num = 0; /* number of backends degeneration requested */
int degenerate_node[MAX_NUM_BACKENDS]; /* degeneration requested backend list */
+ POOL_STATUS status;
POOL_MEMORY_POOL *old_context;
@@ -3572,6 +3634,31 @@ POOL_STATUS read_kind_from_backend(POOL_CONNECTION *frontend, POOL_CONNECTION_PO
memset(kind_map, 0, sizeof(kind_map));
+ if (MASTER_SLAVE)
+ {
+ status = read_kind_from_one_backend(frontend, backend, (char *)&kind, MASTER_NODE_ID);
+ if (status != POOL_CONTINUE)
+ {
+ pool_error("read_kind_from_backend: read_kind_from_one_backend for master node %d failed",
+ MASTER_NODE_ID);
+ return status;
+ }
+
+ /*
+ * 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')
+ {
+ *decided_kind = 'A';
+ pool_log("read_kind_from_backend: received notification message for master node %d",
+ MASTER_NODE_ID);
+ return POOL_CONTINUE;
+ }
+ pool_unread(CONNECTION(backend, MASTER_NODE_ID), &kind, sizeof(kind));
+ }
+
for (i=0;i<NUM_BACKENDS;i++)
{
/* initialize degenerate record */
diff --git a/pool_query_context.c b/pool_query_context.c
index 6dbe397b4..c2ae66b8e 100644
--- a/pool_query_context.c
+++ b/pool_query_context.c
@@ -278,8 +278,14 @@ int pool_virtual_master_db_node_id(void)
}
/*
- * No query context exists. Returns master node id in private buffer.
+ * No query context exists. If in master/slave mode, returns
+ * primary node if exists. Oterwise returns my_master_node_id,
+ * which represents the last REAL_MASTER_NODE_ID.
*/
+ if (MASTER_SLAVE)
+ {
+ return PRIMARY_NODE_ID;
+ }
return my_master_node_id;
}
diff --git a/pool_stream.c b/pool_stream.c
index 30e8f7ac3..49cb927ef 100644
--- a/pool_stream.c
+++ b/pool_stream.c
@@ -851,6 +851,66 @@ int pool_unread(POOL_CONNECTION *cp, void *data, int len)
return 0;
}
+/*
+ * pool_push: Push data into buffer stack.
+ */
+int pool_push(POOL_CONNECTION *cp, void *data, int len)
+{
+ char *p;
+
+ pool_debug("pool_push: len: %d", len);
+
+ if (cp->bufsz3 == 0)
+ {
+ p = cp->buf3 = malloc(len);
+ if (p == NULL)
+ {
+ pool_error("pool_push: malloc failed. len:%d", len);
+ return -1;
+ }
+ }
+ else
+ {
+ p = cp->buf3 + cp->bufsz3;
+ cp->buf3 = realloc(cp->buf3, cp->bufsz3 + len);
+ }
+
+ memcpy(p, data, len);
+ cp->bufsz3 += len;
+
+ return 0;
+}
+
+/*
+ * pool_pop: Pop data from buffer stack and put back data using
+ * pool_unread.
+ */
+void pool_pop(POOL_CONNECTION *cp, int *len)
+{
+ if (cp->bufsz3 == 0)
+ {
+ *len = 0;
+ pool_debug("pool_pop: len: %d", *len);
+ return;
+ }
+
+ pool_unread(cp, cp->buf3, cp->bufsz3);
+ *len = cp->bufsz3;
+ free(cp->buf3);
+ cp->bufsz3 = 0;
+ cp->buf3 = NULL;
+ pool_debug("pool_pop: len: %d", *len);
+}
+
+/*
+ * pool_stacklen: Returns buffer stack length
+ * pool_unread.
+ */
+int pool_stacklen(POOL_CONNECTION *cp)
+{
+ return cp->bufsz3;
+}
+
/*
* set non-block flag
*/
diff --git a/pool_stream.h b/pool_stream.h
index dedd5a1a6..188c1aa4e 100644
--- a/pool_stream.h
+++ b/pool_stream.h
@@ -6,7 +6,7 @@
* pgpool: a language independent connection pool server for PostgreSQL
* written by Tatsuo Ishii
*
- * Copyright (c) 2003-2010 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
@@ -52,6 +52,9 @@ extern int pool_flush_it(POOL_CONNECTION *cp);
extern int pool_write_and_flush(POOL_CONNECTION *cp, void *buf, int len);
extern char *pool_read_string(POOL_CONNECTION *cp, int *len, int line);
extern int pool_unread(POOL_CONNECTION *cp, void *data, int len);
+extern int pool_push(POOL_CONNECTION *cp, void *data, int len);
+extern void pool_pop(POOL_CONNECTION *cp, int *len);
+extern int pool_stacklen(POOL_CONNECTION *cp);
extern void pool_set_nonblock(int fd);
extern void pool_unset_nonblock(int fd);
[parent: 05e02574ea6c]