pgpool2.git / summary / log / commit / refs
commit 450b460ac8fbab9f3f18237fd2f6ed6817c39673
Author: Tatsuo Ishii <ishii at sraoss.co.jp>
Date: Sat Jul 10 11:18:28 2010 +0000
Now transaction state is per backend.
Change TSTATE macro.
Add new INTERNAL_TRANSACTION_STARTED macro.
Made appropreate changes to other modules.
child.c | 2 +-
pool.h | 4 +++-
pool_lobj.c | 2 +-
pool_process_query.c | 42 +++++++++++++++++++++++++-----------------
pool_proto2.c | 16 ++++++++++++----
pool_proto_modules.c | 15 ++++++++++-----
pool_proto_modules.h | 2 +-
pool_query_cache.c | 2 +-
pool_query_context.c | 6 +++---
pool_rewrite_query.c | 4 ++--
10 files changed, 59 insertions(+), 36 deletions(-)
diff --git a/child.c b/child.c
index c3a21ea1f..872cbe000 100644
--- a/child.c
+++ b/child.c
@@ -983,7 +983,7 @@ static bool connect_using_existing_connection(POOL_CONNECTION *frontend,
len = htonl(5);
pool_write(frontend, &len, sizeof(len));
- tstate = TSTATE(backend);
+ tstate = TSTATE(backend, MASTER_NODE_ID);
pool_write(frontend, &tstate, 1);
}
diff --git a/pool.h b/pool.h
index 835ec628f..f32632c65 100644
--- a/pool.h
+++ b/pool.h
@@ -162,6 +162,7 @@ typedef struct {
char tstate; /* transaction state (V3 only) */
+ bool is_internal_transaction_started; /* True if an internal transaction has already started */
/*
* following are used to remember when re-use the authenticated connection
*/
@@ -285,7 +286,8 @@ extern int pool_virtual_master_db_node_id(void);
#define PARALLEL_MODE (pool_config->parallel_mode)
#define RAW_MODE (!REPLICATION && !PARALLEL_MODE && !MASTER_SLAVE)
#define MAJOR(p) MASTER_CONNECTION(p)->sp->major
-#define TSTATE(p) MASTER(p)->tstate
+#define TSTATE(p, i) (CONNECTION(p, i)->tstate)
+#define INTERNAL_TRANSACTION_STARTED(p, i) (CONNECTION(p, i)->is_internal_transaction_started)
#define SYSDB_INFO (system_db_info->info)
#define SYSDB_CONNECTION (system_db_info->connection)
#define SYSDB_STATUS (*system_db_info->system_db_status)
diff --git a/pool_lobj.c b/pool_lobj.c
index 607ac9ff8..3ac1c6d96 100644
--- a/pool_lobj.c
+++ b/pool_lobj.c
@@ -172,7 +172,7 @@ char *pool_rewrite_lo_creat(char kind, char *packet, int packet_len,
/*
* If transaction state is E, do_command failed to execute command
*/
- if (TSTATE(backend) == 'E')
+ if (TSTATE(backend, MASTER_NODE_ID) == 'E')
{
pool_log("pool_check_lo_creat: failed to execute: %s", qbuf);
return NULL;
diff --git a/pool_process_query.c b/pool_process_query.c
index 8d5b66cce..ed0627d11 100644
--- a/pool_process_query.c
+++ b/pool_process_query.c
@@ -154,7 +154,10 @@ POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
else /* no more query(st == 2) */
{
- TSTATE(backend) = 'I';
+ for (i=0;i<NUM_BACKENDS;i++)
+ {
+ TSTATE(backend, i) = 'I';
+ }
frontend->no_forward = 0;
return POOL_CONTINUE;
}
@@ -1476,8 +1479,10 @@ POOL_STATUS SimpleForwardToFrontend(char kind, POOL_CONNECTION *frontend, POOL_C
}
}
}
-
+
+/*
pool_unset_query_in_progress();
+*/
return POOL_CONTINUE;
}
@@ -2015,7 +2020,7 @@ static int reset_backend(POOL_CONNECTION_POOL *backend, int qcnt)
query = pool_config->reset_query_list[qcnt];
/* if transaction state is idle, we don't need to issue ABORT */
- if (TSTATE(backend) == 'I' && !strcmp("ABORT", query))
+ if (TSTATE(backend, MASTER_NODE_ID) == 'I' && !strcmp("ABORT", query))
return 0;
pool_set_timeout(10);
@@ -2038,7 +2043,7 @@ int load_balance_enabled(POOL_CONNECTION_POOL *backend, Node* node, char *sql)
return (pool_config->load_balance_mode &&
(DUAL_MODE || pool_config->parallel_mode) &&
MAJOR(backend) == PROTO_MAJOR_V3 &&
- TSTATE(backend) == 'I' &&
+ TSTATE(backend, MASTER_NODE_ID) == 'I' &&
is_select_query(node, sql) &&
!is_sequence_query(node));
}
@@ -4131,21 +4136,22 @@ static bool is_internal_transaction_needed(Node *node)
return false;
}
+/*
+ * Start an internal transaction if necessary.
+ */
POOL_STATUS start_internal_transaction(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend, Node *node)
{
int i;
- if (TSTATE(backend) != 'I')
- return POOL_CONTINUE;
-
- /* if we are not in a transaction block,
+ /* If we are not in a transaction block,
* start a new transaction
*/
if (is_internal_transaction_needed(node))
{
for (i=0;i<NUM_BACKENDS;i++)
{
- if (VALID_BACKEND(i))
+ if (VALID_BACKEND(i) && !INTERNAL_TRANSACTION_STARTED(backend, i) &&
+ TSTATE(backend, i) == 'I')
{
per_node_statement_log(backend, i, "BEGIN");
@@ -4153,15 +4159,16 @@ POOL_STATUS start_internal_transaction(POOL_CONNECTION *frontend, POOL_CONNECTIO
MASTER_CONNECTION(backend)->pid, MASTER_CONNECTION(backend)->key, 0) != POOL_CONTINUE)
return POOL_END;
}
+ /* Mark that we started new transaction */
+ INTERNAL_TRANSACTION_STARTED(backend, i) = true;
}
-
- /* mark that we started new transaction */
- internal_transaction_started = 1;
}
return POOL_CONTINUE;
}
-
+/*
+ * End internal transaction.
+ */
POOL_STATUS end_internal_transaction(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend)
{
int i;
@@ -4180,7 +4187,8 @@ POOL_STATUS end_internal_transaction(POOL_CONNECTION *frontend, POOL_CONNECTION_
/* We need to commit from secondary to master. */
for (i=0;i<NUM_BACKENDS;i++)
{
- if (VALID_BACKEND(i) && !IS_MASTER_NODE_ID(i))
+ if (VALID_BACKEND(i) && !IS_MASTER_NODE_ID(i) &&
+ TSTATE(backend, i) == 'I')
{
per_node_statement_log(backend, i, "COMMIT");
@@ -4188,7 +4196,7 @@ POOL_STATUS end_internal_transaction(POOL_CONNECTION *frontend, POOL_CONNECTION_
if (do_command(frontend, CONNECTION(backend, i), "COMMIT", MAJOR(backend),
MASTER_CONNECTION(backend)->pid, MASTER_CONNECTION(backend)->key, 1) != POOL_CONTINUE)
{
- internal_transaction_started = 0;
+ INTERNAL_TRANSACTION_STARTED(backend, i) = true;
POOL_SETMASK(&oldmask);
return POOL_END;
}
@@ -4200,12 +4208,12 @@ POOL_STATUS end_internal_transaction(POOL_CONNECTION *frontend, POOL_CONNECTION_
if (do_command(frontend, MASTER(backend), "COMMIT", MAJOR(backend),
MASTER_CONNECTION(backend)->pid, MASTER_CONNECTION(backend)->key, 1) != POOL_CONTINUE)
{
- internal_transaction_started = 0;
+ INTERNAL_TRANSACTION_STARTED(backend, MASTER_NODE_ID) = true;
POOL_SETMASK(&oldmask);
return POOL_END;
}
- internal_transaction_started = 0;
+ INTERNAL_TRANSACTION_STARTED(backend, MASTER_NODE_ID) = true;
POOL_SETMASK(&oldmask);
return POOL_CONTINUE;
}
diff --git a/pool_proto2.c b/pool_proto2.c
index f2ec2d169..1e73cfde1 100644
--- a/pool_proto2.c
+++ b/pool_proto2.c
@@ -385,10 +385,18 @@ POOL_STATUS ErrorResponse(POOL_CONNECTION *frontend,
return POOL_END;
/* change transaction state */
- if (TSTATE(backend) == 'T')
- TSTATE(backend) = 'E';
- else
- TSTATE(backend) = 'I';
+ for (i=0;i<NUM_BACKENDS;i++)
+ {
+ if (VALID_BACKEND(i))
+ {
+ if (TSTATE(backend, i) == 'T')
+ TSTATE(backend, i) = 'E';
+ }
+ else
+ {
+ TSTATE(backend, i) = 'I';
+ }
+ }
return POOL_CONTINUE;
}
diff --git a/pool_proto_modules.c b/pool_proto_modules.c
index 318ed8b82..4f7633e19 100644
--- a/pool_proto_modules.c
+++ b/pool_proto_modules.c
@@ -398,7 +398,12 @@ static int is_temp_table(POOL_CONNECTION_POOL *backend, Node *node);
if (MAJOR(backend) == PROTO_MAJOR_V2 && is_start_transaction_query(node))
{
- TSTATE(backend) = 'T';
+ int i;
+
+ for (i=0;i<NUM_BACKENDS;i++)
+ {
+ TSTATE(backend, i) = 'T';
+ }
}
if (!RAW_MODE)
@@ -550,7 +555,7 @@ POOL_STATUS Execute(POOL_CONNECTION *frontend,
if ((IsA(node, PrepareStmt) || IsA(node, DeallocateStmt) ||
IsA(node, VariableSetStmt)) &&
- MASTER_SLAVE && TSTATE(backend) != 'E')
+ MASTER_SLAVE && TSTATE(backend, MASTER_NODE_ID) != 'E')
{
/*
* PREPARE, DEALLOCATE, SET, DISCARD
@@ -826,7 +831,7 @@ POOL_STATUS Parse(POOL_CONNECTION *frontend,
{
char kind;
- if (TSTATE(backend) != 'T')
+ if (TSTATE(backend, MASTER_NODE_ID) != 'T')
{
/* synchronize transaction state */
for (i = 0; i < NUM_BACKENDS; i++)
@@ -1318,7 +1323,7 @@ POOL_STATUS ProcessFrontendResponse(POOL_CONNECTION *frontend,
allow_close_transaction = 0;
if (MASTER_SLAVE &&
- (TSTATE(backend) != 'I' || receive_extended_begin))
+ (TSTATE(backend, MASTER_NODE_ID) != 'I' || receive_extended_begin))
{
pool_debug("kind: %c master_slave_dml enabled", fkind);
master_slave_was_enabled = 1;
@@ -1339,7 +1344,7 @@ POOL_STATUS ProcessFrontendResponse(POOL_CONNECTION *frontend,
fkind == 'C' || fkind == 'B' || fkind == 'F' || fkind == 'd' || fkind == 'c'))
{
if (MASTER_SLAVE &&
- (TSTATE(backend) != 'I' || receive_extended_begin))
+ (TSTATE(backend, MASTER_NODE_ID) != 'I' || receive_extended_begin))
{
pool_debug("kind: %c master_slave_dml enabled", fkind);
master_slave_was_enabled = 1;
diff --git a/pool_proto_modules.h b/pool_proto_modules.h
index b38565f8f..f40c34f35 100644
--- a/pool_proto_modules.h
+++ b/pool_proto_modules.h
@@ -32,11 +32,11 @@
#include "parser/pg_list.h"
#include "parser/parsenodes.h"
#include "pool_rewrite_query.h"
+#include "pool_session_context.h"
#define SPECIFIED_ERROR 1
#define POOL_ERROR_QUERY "send invalid query from pgpool to abort transaction"
-
extern int force_replication;
extern int replication_was_enabled; /* replication mode was enabled */
extern int master_slave_was_enabled; /* master/slave mode was enabled */
diff --git a/pool_query_cache.c b/pool_query_cache.c
index fb42efd63..955076bbb 100644
--- a/pool_query_cache.c
+++ b/pool_query_cache.c
@@ -1004,7 +1004,7 @@ POOL_STATUS pool_execute_query_cache_lookup(POOL_CONNECTION *frontend, POOL_CONN
}
- if (pool_query_cache_lookup(frontend, parsed_query, backend->info->database, TSTATE(backend)) == POOL_CONTINUE)
+ if (pool_query_cache_lookup(frontend, parsed_query, backend->info->database, TSTATE(backend, MASTER_NODE_ID)) == POOL_CONTINUE)
{
free(parsed_query);
parsed_query = NULL;
diff --git a/pool_query_context.c b/pool_query_context.c
index 8bd444cc7..f3e3516f8 100644
--- a/pool_query_context.c
+++ b/pool_query_context.c
@@ -267,7 +267,7 @@ void pool_where_to_send(POOL_QUERY_CONTEXT *query_context, char *query, Node *no
MAJOR(backend) == PROTO_MAJOR_V3)
{
/* Outside of an explicit transaction? */
- if (TSTATE(backend) == 'I')
+ if (TSTATE(backend, MASTER_NODE_ID) == 'I')
{
BackendInfo *bkinfo = pool_get_node_info(session_context->load_balance_node_id);
@@ -326,7 +326,7 @@ void pool_where_to_send(POOL_QUERY_CONTEXT *query_context, char *query, Node *no
{
if (pool_config->load_balance_mode &&
MAJOR(backend) == PROTO_MAJOR_V3 &&
- TSTATE(backend) == 'I' &&
+ TSTATE(backend, MASTER_NODE_ID) == 'I' &&
is_select_query(node, query) &&
!is_sequence_query(node))
{
@@ -346,7 +346,7 @@ void pool_where_to_send(POOL_QUERY_CONTEXT *query_context, char *query, Node *no
{
if (pool_config->load_balance_mode &&
MAJOR(backend) == PROTO_MAJOR_V3 &&
- TSTATE(backend) == 'I' &&
+ TSTATE(backend, MASTER_NODE_ID) == 'I' &&
is_select_query(node, query) &&
!is_sequence_query(node))
{
diff --git a/pool_rewrite_query.c b/pool_rewrite_query.c
index cc7866298..c13687db0 100644
--- a/pool_rewrite_query.c
+++ b/pool_rewrite_query.c
@@ -389,7 +389,7 @@ RewriteQuery *rewrite_query_stmt(Node *node,POOL_CONNECTION *frontend,POOL_CONNE
}
else
{
- if(TSTATE(backend) == 'T' &&
+ if(TSTATE(backend, MASTER_NODE_ID) == 'T' &&
message->r_code == SELECT_RELATION_ERROR)
{
/*
@@ -440,7 +440,7 @@ RewriteQuery *rewrite_query_stmt(Node *node,POOL_CONNECTION *frontend,POOL_CONNE
message->rewrite_query, "", __FILE__,
__LINE__);
- if(TSTATE(backend) == 'T')
+ if(TSTATE(backend, MASTER_NODE_ID) == 'T')
{
/* In Transaction, send the invalid message to backend to abort this transaction */
pool_debug("rewrite_query_stmt(insert): Inside transaction. Abort transaction");
[parent: f80199fc887c]