pgpool2.git / summary / log / commit / refs
commit 3c2962000e6971eefadaf53f6947a90d139cf754
Author: Yoshiyuki Asaba <y-asaba at pgfoundry.org>
Date: Tue Jul 10 09:56:07 2007 +0000
* Change type of InRecovery variable to sig_atomic_t*
* Fixed kind mismatch error when deadlock error occured.
The problem is the following senario.
A: BEGIN:
B: BEGIN;
A: LOCK TABLE t1 IN SHARE ROW EXCLUSIVE MODE;
B: LOCK TABLE t2 IN SHARE ROW EXCLUSIVE MODE;
A: LOCK TABLE t2 IN SHARE ROW EXCLUSIVE MODE;
B: LOCK TABLE t1 IN SHARE ROW EXCLUSIVE MODE;
Transaction "A" aborts on master node, but it completes on another
nodes. It causes wrong failover.
So pgpool checks deadlock error(code == '40P01') and sends error query
to another nodes.
main.c | 2 +-
pool.h | 6 ++-
pool_process_query.c | 133 +++++++++++++++++++++++++++++++++++++++++++++++++--
pool_stream.c | 28 +++++++++++
4 files changed, 163 insertions(+), 6 deletions(-)
diff --git a/main.c b/main.c
index d45a8b467..e63187338 100644
--- a/main.c
+++ b/main.c
@@ -127,7 +127,7 @@ static int stop_sig = SIGTERM; /* stopping signal default value */
static int health_check_timer_expired; /* non 0 if health check timer expired */
POOL_REQUEST_INFO *Req_info; /* request info area in shared memory */
-volatile int *InRecovery; /* non 0 if recovery is started */
+volatile sig_atomic_t *InRecovery; /* non 0 if recovery is started */
static volatile sig_atomic_t failover_request = 0;
static volatile sig_atomic_t sigchld_request = 0;
static volatile sig_atomic_t wakeup_request = 0;
diff --git a/pool.h b/pool.h
index 4f51931b0..32c8d9b9e 100644
--- a/pool.h
+++ b/pool.h
@@ -351,7 +351,8 @@ typedef enum SemNum
typedef enum {
NODE_UP_REQUEST = 0,
NODE_DOWN_REQUEST,
- NODE_RECOVERY_REQUEST
+ NODE_RECOVERY_REQUEST,
+ CLOSE_IDLE_REQUEST
} POOL_REQUEST_KIND;
typedef struct {
@@ -374,7 +375,7 @@ extern int in_load_balance; /* non 0 if in load balance mode */
extern int selected_slot; /* selected DB node for load balance */
extern int master_slave_dml; /* non 0 if master/slave mode is specified in config file */
extern POOL_REQUEST_INFO *Req_info;
-extern volatile int *InRecovery;
+extern volatile sig_atomic_t *InRecovery;
extern char remote_ps_data[]; /* used for set_ps_display */
/*
@@ -401,6 +402,7 @@ extern int pool_flush(POOL_CONNECTION *cp);
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_do_auth(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend);
extern int pool_do_reauth(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *cp);
diff --git a/pool_process_query.c b/pool_process_query.c
index 493793dc6..3ae00add8 100644
--- a/pool_process_query.c
+++ b/pool_process_query.c
@@ -50,6 +50,9 @@
#define INIT_STATEMENT_LIST_SIZE 8
+#define DEADLOCK_ERROR_CODE "40P01"
+#define POOL_ERROR_QUERY "send invalid query from pgpool to abort transaction"
+
typedef struct {
char *portal_name;
Node *stmt;
@@ -194,6 +197,7 @@ static void query_cache_register(char kind, POOL_CONNECTION *frontend, char *dat
static POOL_STATUS start_internal_transaction(POOL_CONNECTION_POOL *backend, Node *node);
static POOL_STATUS end_internal_transaction(POOL_CONNECTION_POOL *backend);
static int extract_ntuples(char *message);
+static int detect_deadlock_error(POOL_CONNECTION *master, int major);
POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
@@ -763,6 +767,7 @@ static POOL_STATUS SimpleQuery(POOL_CONNECTION *frontend,
List *parse_tree_list;
Node *node = NULL, *node1;
POOL_STATUS status;
+ int deadlock_detected = 0;
if (query == NULL) /* need to read query from frontend? */
{
@@ -1045,6 +1050,21 @@ static POOL_STATUS SimpleQuery(POOL_CONNECTION *frontend,
if (wait_for_query_response(MASTER(backend), string) != POOL_CONTINUE)
return POOL_END;
+
+ /*
+ * We must check deadlock error because a aborted transaction
+ * by detecting deadlock isn't same on all nodes.
+ * If a transaction is aborted on master node, pgpool send a
+ * error query to another nodes.
+ */
+ deadlock_detected = detect_deadlock_error(MASTER(backend), MAJOR(backend));
+ if (deadlock_detected < 0)
+ return POOL_END;
+ }
+ if (deadlock_detected)
+ {
+ string = POOL_ERROR_QUERY;
+ len = strlen(string) + 1;
}
/* send query to other nodes */
@@ -1137,6 +1157,7 @@ static POOL_STATUS Execute(POOL_CONNECTION *frontend,
Portal *portal;
char *string1;
PrepareStmt *p_stmt;
+ int deadlock_detected = 0;
/* read Execute packet */
if (pool_read(frontend, &len, sizeof(len)) < 0)
@@ -1198,6 +1219,16 @@ static POOL_STATUS Execute(POOL_CONNECTION *frontend,
if (synchronize(CONNECTION(backend, MASTER_NODE_ID)))
return POOL_END;
}
+
+ /*
+ * We must check deadlock error because a aborted transaction
+ * by detecting deadlock isn't same on all nodes.
+ * If a transaction is aborted on master node, pgpool send a
+ * error query to another nodes.
+ */
+ deadlock_detected = detect_deadlock_error(MASTER(backend), MAJOR(backend));
+ if (deadlock_detected < 0)
+ return POOL_END;
}
/* send query to other nodes */
@@ -1205,8 +1236,15 @@ static POOL_STATUS Execute(POOL_CONNECTION *frontend,
{
if (!VALID_BACKEND(i) || IS_MASTER_NODE_ID(i))
continue;
-
- if (send_execute_message(backend, MASTER_NODE_ID, len, string) != POOL_CONTINUE)
+ if (deadlock_detected)
+ {
+ if (send_simplequery_message(CONNECTION(backend, i),
+ strlen(POOL_ERROR_QUERY)+1,
+ POOL_ERROR_QUERY,
+ MAJOR(backend)))
+ return POOL_END;
+ }
+ else if (send_execute_message(backend, i, len, string) != POOL_CONTINUE)
return POOL_END;
}
@@ -4023,7 +4061,7 @@ static POOL_STATUS do_command(POOL_CONNECTION *backend, char *query, int protoMa
*/
static POOL_STATUS do_error_command(POOL_CONNECTION *backend, int major)
{
- char *error_query = "send invalid query from pgpool to abort transaction";
+ char *error_query = POOL_ERROR_QUERY;
int status, len;
char kind;
char *string;
@@ -4748,3 +4786,92 @@ static int extract_ntuples(char *message)
return atoi(rows);
}
+
+static int detect_deadlock_error(POOL_CONNECTION *master, int major)
+{
+ int deadlock = 0;
+ char kind;
+ int readlen = 0, len;
+ char *buf;
+ char *p, *str;
+
+ if ((buf = malloc(1024)) == NULL)
+ {
+ pool_error("detect_deadlock_error: malloc failed");
+ return -1;
+ }
+
+ if (pool_read(master, &kind, sizeof(kind)))
+ return POOL_END;
+ readlen += sizeof(kind);
+ p = buf;
+ memcpy(p, &kind, sizeof(kind));
+ p += sizeof(kind);
+
+ if (kind == 'E') /* deadlock error? */
+ {
+ pool_debug("error");
+ /* read actual query */
+ if (major == PROTO_MAJOR_V3)
+ {
+ char *error_code;
+
+ if (pool_read(master, &len, sizeof(len)) < 0)
+ return POOL_END;
+ readlen += sizeof(len);
+ memcpy(p, &len, sizeof(len));
+ p += sizeof(len);
+
+ len = ntohl(len) - 4;
+ str = malloc(len);
+ pool_read(master, str, len);
+ readlen += len;
+ if (readlen > 1024)
+ {
+ buf = realloc(buf, readlen);
+ if (buf == NULL)
+ {
+ pool_error("detect_deadlock_error: malloc failed");
+ return -1;
+ }
+ }
+ memcpy(p, str, len);
+
+ error_code = str;
+ while (*error_code)
+ {
+ if (*error_code == 'C')
+ {
+ if (strcmp(error_code+1, DEADLOCK_ERROR_CODE) == 0) /* deadlock error */
+ {
+ pool_debug("SimpleQuery: receive deadlock error from master node.");
+ deadlock = 1;
+ }
+ break;
+ }
+ else
+ error_code = error_code + strlen(error_code) + 1;
+ }
+ free(str);
+ }
+ else
+ {
+ str = pool_read_string(master, &len, 0);
+ readlen += len;
+ if (readlen > 1024)
+ {
+ buf = realloc(buf, readlen);
+ if (buf == NULL)
+ {
+ pool_error("detect_deadlock_error: malloc failed");
+ return -1;
+ }
+ }
+ memcpy(p, str, len);
+ }
+ }
+ if (pool_unread(master, buf, readlen) != 0)
+ deadlock = -1;
+ free(buf);
+ return deadlock;
+}
diff --git a/pool_stream.c b/pool_stream.c
index 09f67d310..1fd0d145c 100644
--- a/pool_stream.c
+++ b/pool_stream.c
@@ -754,3 +754,31 @@ static int consume_pending_data(POOL_CONNECTION *cp, void *data, int len)
return consume_size;
}
+
+/*
+ * pool_unread: Put back data to input buffer
+ */
+int pool_unread(POOL_CONNECTION *cp, void *data, int len)
+{
+ void *p = cp->hp;
+ int n = cp->len + len;
+ int realloc_size;
+
+ if (cp->bufsz < n)
+ {
+ realloc_size = (n/READBUFSZ+1)*READBUFSZ;
+ p = realloc(cp->hp, realloc_size);
+ if (p == NULL)
+ {
+ pool_error("pool_unread: realloc failed");
+ return -1;
+ }
+ cp->hp = p;
+ }
+ if (cp->len != 0)
+ memmove(p + len, cp->hp + cp->po, cp->len);
+ memmove(p, data, len);
+ cp->len = n;
+ cp->po = 0;
+ return 0;
+}
[parent: 8db15261d8ed]