pgpool2.git / summary / log / commit / refs
commit bcd103870b0dbcd8be535193c6ca8d50f60b2f98
Author: Yoshiyuki Asaba <y-asaba at pgfoundry.org>
Date: Tue Aug 28 02:28:20 2007 +0000
* Fix handling Parse message.
Parse message including INSERT/UPDATE/DELETE query acquires
RowExclusiveLock. However pgpool did not wait a response from each
node. It was possible to occur deadlock problem.
* Fix memory leak in reset_prepared_list()
pool_process_query.c | 214 +++++++++++++++++++++++++++++++++++----------------
1 file changed, 148 insertions(+), 66 deletions(-)
diff --git a/pool_process_query.c b/pool_process_query.c
index 24afa9fe8..bb3792b71 100644
--- a/pool_process_query.c
+++ b/pool_process_query.c
@@ -80,6 +80,9 @@ static POOL_STATUS SimpleQuery(POOL_CONNECTION *frontend,
static POOL_STATUS Execute(POOL_CONNECTION *frontend,
POOL_CONNECTION_POOL *backend);
+static POOL_STATUS Parse(POOL_CONNECTION *frontend,
+ POOL_CONNECTION_POOL *backend);
+
#ifdef NOT_USED
static POOL_STATUS Sync(POOL_CONNECTION *frontend,
POOL_CONNECTION_POOL *backend);
@@ -128,6 +131,9 @@ static POOL_STATUS ProcessFrontendResponse(POOL_CONNECTION *frontend,
static POOL_STATUS send_simplequery_message(POOL_CONNECTION *backend, int len, char *string, int major);
static POOL_STATUS wait_for_query_response(POOL_CONNECTION *backend, char *string);
+static POOL_STATUS send_extended_protocol_message(POOL_CONNECTION_POOL *backend,
+ int node_id, char *kind,
+ int len, char *string);
static POOL_STATUS send_execute_message(POOL_CONNECTION_POOL *backend,
int node_id, int len, char *string);
static int synchronize(POOL_CONNECTION *cp);
@@ -1373,14 +1379,19 @@ static POOL_STATUS Execute(POOL_CONNECTION *frontend,
return POOL_CONTINUE;
}
-static POOL_STATUS send_execute_message(POOL_CONNECTION_POOL *backend,
- int node_id, int len, char *string)
+
+/*
+ * Extended query protocol has to send Flush message.
+ */
+static POOL_STATUS send_extended_protocol_message(POOL_CONNECTION_POOL *backend,
+ int node_id, char *kind,
+ int len, char *string)
{
POOL_CONNECTION *cp = CONNECTION(backend, node_id);
int sendlen;
/* forward the query to the backend */
- pool_write(cp, "E", 1);
+ pool_write(cp, kind, 1);
sendlen = htonl(len + 4);
pool_write(cp, &sendlen, sizeof(sendlen));
pool_write(cp, string, len);
@@ -1399,6 +1410,127 @@ static POOL_STATUS send_execute_message(POOL_CONNECTION_POOL *backend,
return POOL_CONTINUE;
}
+static POOL_STATUS send_execute_message(POOL_CONNECTION_POOL *backend,
+ int node_id, int len, char *string)
+{
+ return send_extended_protocol_message(backend, node_id, "E", len, string);
+}
+
+
+/*
+ * process Parse (V3 only)
+ */
+static POOL_STATUS Parse(POOL_CONNECTION *frontend,
+ POOL_CONNECTION_POOL *backend)
+{
+ int len;
+ char *string;
+ int i;
+ Portal *portal;
+ POOL_MEMORY_POOL *old_context;
+ PrepareStmt *p_stmt;
+ char *name, *stmt;
+ List *parse_tree_list;
+ Node *node = NULL;
+
+ /* read Parse packet */
+ if (pool_read(frontend, &len, sizeof(len)) < 0)
+ return POOL_END;
+
+ len = ntohl(len) - 4;
+ string = pool_read2(frontend, len);
+
+ pool_debug("Parse: portal name <%s>", string);
+
+ name = string;
+ stmt = string + strlen(string) + 1;
+
+ parse_tree_list = raw_parser(stmt);
+ if (parse_tree_list == NIL)
+ {
+ free_parser();
+ }
+ else
+ {
+ node = (Node *) lfirst(list_head(parse_tree_list));
+
+ if (prepare_memory_context == NULL)
+ {
+ prepare_memory_context = pool_memory_create();
+ if (prepare_memory_context == NULL)
+ {
+ pool_error("Simple Query: pool_memory_create() failed");
+ return POOL_ERROR;
+ }
+ }
+ /* switch memory context */
+ old_context = pool_memory;
+ pool_memory = prepare_memory_context;
+
+ portal = malloc(sizeof(Portal));
+ /* translate Parse message to PrepareStmt */
+ p_stmt = palloc(sizeof(PrepareStmt));
+ p_stmt->type = T_PrepareStmt;
+ p_stmt->name = pstrdup(name);
+ p_stmt->query = copyObject(node);
+ portal->stmt = (Node *)p_stmt;
+ portal->portal_name = NULL;
+
+ if (*name)
+ {
+ pending_function = add_prepared_list;
+ pending_prepared_portal = portal;
+ }
+ else /* unnamed statement */
+ {
+ pending_function = add_unnamed_portal;
+ pfree(p_stmt->name);
+ p_stmt->name = NULL;
+ pending_prepared_portal = portal;
+ }
+
+ /* switch old memory context */
+ pool_memory = old_context;
+ free_parser();
+ }
+
+ /* send to master node */
+ if (send_extended_protocol_message(backend, MASTER_NODE_ID,
+ "P", len, string))
+ return POOL_END;
+
+ if (REPLICATION || PARALLEL_MODE)
+ {
+ /* We must synchronize because Parse message acquires table
+ * locks.
+ */
+ if (pool_config->replication_strict)
+ {
+ pool_debug("waiting for master completing the query");
+ if (synchronize(MASTER(backend)))
+ return POOL_END;
+ }
+
+ for (i=0;i<NUM_BACKENDS;i++)
+ {
+ if (VALID_BACKEND(i) && !IS_MASTER_NODE_ID(i))
+ {
+ if (send_extended_protocol_message(backend, i,
+ "P", len, string))
+ return POOL_END;
+
+ if (pool_config->replication_strict)
+ {
+ pool_debug("waiting for master completing the query");
+ if (synchronize(CONNECTION(backend, i)))
+ return POOL_END;
+ }
+ }
+ }
+ }
+ return POOL_CONTINUE;
+}
+
#ifdef NOT_USED
/*
* process Sync (V3 only)
@@ -2577,6 +2709,10 @@ static POOL_STATUS ProcessFrontendResponse(POOL_CONNECTION *frontend,
status = Execute(frontend, backend);
break;
+ case 'P':
+ status = Parse(frontend, backend);
+ break;
+
default:
if (MAJOR(backend) == PROTO_MAJOR_V3)
{
@@ -3533,68 +3669,7 @@ POOL_STATUS SimpleForwardToBackend(char kind, POOL_CONNECTION *frontend, POOL_CO
}
}
- if (kind == 'P') /* Parse message */
- {
- Portal *portal;
- POOL_MEMORY_POOL *old_context;
- PrepareStmt *p_stmt;
- char *name, *stmt;
- List *parse_tree_list;
- Node *node = NULL;
-
- name = p;
- stmt = p + strlen(p) + 1;
-
- parse_tree_list = raw_parser(stmt);
- if (parse_tree_list == NIL)
- {
- free_parser();
- }
- else
- {
- node = (Node *) lfirst(list_head(parse_tree_list));
-
- if (prepare_memory_context == NULL)
- {
- prepare_memory_context = pool_memory_create();
- if (prepare_memory_context == NULL)
- {
- pool_error("Simple Query: pool_memory_create() failed");
- return POOL_ERROR;
- }
- }
- /* switch memory context */
- old_context = pool_memory;
- pool_memory = prepare_memory_context;
-
- portal = malloc(sizeof(Portal));
- /* translate Parse message to PrepareStmt */
- p_stmt = palloc(sizeof(PrepareStmt));
- p_stmt->type = T_PrepareStmt;
- p_stmt->name = pstrdup(name);
- p_stmt->query = copyObject(node);
- portal->stmt = (Node *)p_stmt;
- portal->portal_name = NULL;
-
- if (*name)
- {
- pending_function = add_prepared_list;
- pending_prepared_portal = portal;
- }
- else /* unnamed statement */
- {
- pending_function = add_unnamed_portal;
- pfree(p_stmt->name);
- p_stmt->name = NULL;
- pending_prepared_portal = portal;
- }
-
- /* switch old memory context */
- pool_memory = old_context;
- free_parser();
- }
- }
- else if (kind == 'B') /* Bind message */
+ if (kind == 'B') /* Bind message */
{
Portal *portal = NULL;
char *stmt_name, *portal_name;
@@ -3634,7 +3709,7 @@ POOL_STATUS SimpleForwardToBackend(char kind, POOL_CONNECTION *frontend, POOL_CO
pending_prepared_portal = NULL;
}
- if (kind == 'P' || kind == 'B' || kind == 'D' || kind == 'C')
+ if (kind == 'B' || kind == 'D' || kind == 'C')
{
int i;
int kind1;
@@ -4617,8 +4692,15 @@ static void del_prepared_list(PreparedStatementList *p, Portal *portal)
static void reset_prepared_list(PreparedStatementList *p)
{
+ int i;
+
if (prepare_memory_context)
{
+ for (i = 0; i < p->cnt; i++)
+ {
+ free(p->portal_list[i]->portal_name);
+ free(p->portal_list[i]);
+ }
pool_memory_delete(prepare_memory_context);
prepare_memory_context = NULL;
free(unnamed_statement);
[parent: 0e3d6d258a83]