Skip to content

M
MariaDB
Commit a0725e13 authored by unknown's avatar unknown
Browse files
Options

BUG#12162 - one can start two transactions with the same XID. 

Now we keep all active XID's in a hash
parent 0bd7011e
Show whitespace changes
Inline Side-by-side
Showing with 211 additions and 93 deletions
mysql-test/r/xa.result
View file @ a0725e13
...@@ -22,6 +22,8 @@ a ...@@ -22,6 +22,8 @@ a
xa start 'testa','testb'; xa start 'testa','testb';
insert t1 values (30); insert t1 values (30);
xa end 'testa','testb'; xa end 'testa','testb';
xa start 'testa','testb';
ERROR XAE08: XAER_DUPID: The XID already exists
xa start 0x7465737462, 0x2030405060, 0xb; xa start 0x7465737462, 0x2030405060, 0xb;
insert t1 values (40); insert t1 values (40);
xa end 'testb',' 0@P`',11; xa end 'testb',' 0@P`',11;
...@@ -35,11 +37,11 @@ formatID gtrid_length bqual_length data ...@@ -35,11 +37,11 @@ formatID gtrid_length bqual_length data
11 5 5 testb 0@P` 11 5 5 testb 0@P`
1 5 5 testatestb 1 5 5 testatestb
xa commit 'testb',0x2030405060,11; xa commit 'testb',0x2030405060,11;
ERROR XAE04: XAER_NOTA: Unknown XID
xa rollback 'testa','testb'; xa rollback 'testa','testb';
xa start 'zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz'; xa start 'zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz';
ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near '' at line 1 ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near '' at line 1
select * from t1; select * from t1;
a a
20 20
40
drop table t1; drop table t1;
mysql-test/t/xa.test
View file @ a0725e13
...@@ -31,6 +31,9 @@ xa end 'testa','testb'; ...@@ -31,6 +31,9 @@ xa end 'testa','testb';
connect (con1,localhost,,,); connect (con1,localhost,,,);
connection con1; connection con1;
--error 1438
xa start 'testa','testb';
# gtrid [ , bqual [ , formatID ] ] # gtrid [ , bqual [ , formatID ] ]
xa start 0x7465737462, 0x2030405060, 0xb; xa start 0x7465737462, 0x2030405060, 0xb;
insert t1 values (40); insert t1 values (40);
...@@ -47,6 +50,7 @@ xa prepare 'testa','testb'; ...@@ -47,6 +50,7 @@ xa prepare 'testa','testb';
xa recover; xa recover;
--error 1397
xa commit 'testb',0x2030405060,11; xa commit 'testb',0x2030405060,11;
xa rollback 'testa','testb'; xa rollback 'testa','testb';
......
sql/ha_innodb.cc
View file @ a0725e13
...@@ -7037,7 +7037,7 @@ innobase_xa_prepare( ...@@ -7037,7 +7037,7 @@ innobase_xa_prepare(
return(0); return(0);
} }
trx->xid=thd->transaction.xid; trx->xid=thd->transaction.xid_state.xid;
/* Release a possible FIFO ticket and search latch. Since we will /* Release a possible FIFO ticket and search latch. Since we will
reserve the kernel mutex, we have to release the search system latch reserve the kernel mutex, we have to release the search system latch
......
sql/handler.cc
View file @ a0725e13
...@@ -547,8 +547,8 @@ void trans_register_ha(THD *thd, bool all, handlerton *ht_arg) ...@@ -547,8 +547,8 @@ void trans_register_ha(THD *thd, bool all, handlerton *ht_arg)
trans->ht[trans->nht++]=ht_arg; trans->ht[trans->nht++]=ht_arg;
DBUG_ASSERT(*ht == ht_arg); DBUG_ASSERT(*ht == ht_arg);
trans->no_2pc|=(ht_arg->prepare==0); trans->no_2pc|=(ht_arg->prepare==0);
if (thd->transaction.xid.is_null()) if (thd->transaction.xid_state.xid.is_null())
thd->transaction.xid.set(thd->query_id); thd->transaction.xid_state.xid.set(thd->query_id);
DBUG_VOID_RETURN; DBUG_VOID_RETURN;
} }
...@@ -595,7 +595,7 @@ int ha_commit_trans(THD *thd, bool all) ...@@ -595,7 +595,7 @@ int ha_commit_trans(THD *thd, bool all)
THD_TRANS *trans= all ? &thd->transaction.all : &thd->transaction.stmt; THD_TRANS *trans= all ? &thd->transaction.all : &thd->transaction.stmt;
bool is_real_trans= all || thd->transaction.all.nht == 0; bool is_real_trans= all || thd->transaction.all.nht == 0;
handlerton **ht= trans->ht; handlerton **ht= trans->ht;
my_xid xid= thd->transaction.xid.get_my_xid(); my_xid xid= thd->transaction.xid_state.xid.get_my_xid();
DBUG_ENTER("ha_commit_trans"); DBUG_ENTER("ha_commit_trans");
if (thd->in_sub_stmt) if (thd->in_sub_stmt)
...@@ -695,7 +695,7 @@ int ha_commit_one_phase(THD *thd, bool all) ...@@ -695,7 +695,7 @@ int ha_commit_one_phase(THD *thd, bool all)
trans->nht=0; trans->nht=0;
trans->no_2pc=0; trans->no_2pc=0;
if (is_real_trans) if (is_real_trans)
thd->transaction.xid.null(); thd->transaction.xid_state.xid.null();
if (all) if (all)
{ {
#ifdef HAVE_QUERY_CACHE #ifdef HAVE_QUERY_CACHE
...@@ -751,7 +751,7 @@ int ha_rollback_trans(THD *thd, bool all) ...@@ -751,7 +751,7 @@ int ha_rollback_trans(THD *thd, bool all)
trans->nht=0; trans->nht=0;
trans->no_2pc=0; trans->no_2pc=0;
if (is_real_trans) if (is_real_trans)
thd->transaction.xid.null(); thd->transaction.xid_state.xid.null();
if (all) if (all)
{ {
thd->variables.tx_isolation=thd->session_tx_isolation; thd->variables.tx_isolation=thd->session_tx_isolation;
...@@ -945,6 +945,7 @@ int ha_recover(HASH *commit_list) ...@@ -945,6 +945,7 @@ int ha_recover(HASH *commit_list)
char buf[XIDDATASIZE*4+6]; // see xid_to_str char buf[XIDDATASIZE*4+6]; // see xid_to_str
sql_print_information("ignore xid %s", xid_to_str(buf, list+i)); sql_print_information("ignore xid %s", xid_to_str(buf, list+i));
#endif #endif
xid_cache_insert(list+i, XA_PREPARED);
found_foreign_xids++; found_foreign_xids++;
continue; continue;
} }
...@@ -1008,10 +1009,8 @@ bool mysql_xa_recover(THD *thd) ...@@ -1008,10 +1009,8 @@ bool mysql_xa_recover(THD *thd)
{ {
List<Item> field_list; List<Item> field_list;
Protocol *protocol= thd->protocol; Protocol *protocol= thd->protocol;
handlerton **ht= handlertons, **end_ht=ht+total_ha; int i=0;
bool error=TRUE; XID_STATE *xs;
int len, got;
XID *list=0;
DBUG_ENTER("mysql_xa_recover"); DBUG_ENTER("mysql_xa_recover");
field_list.push_back(new Item_int("formatID",0,11)); field_list.push_back(new Item_int("formatID",0,11));
...@@ -1021,48 +1020,30 @@ bool mysql_xa_recover(THD *thd) ...@@ -1021,48 +1020,30 @@ bool mysql_xa_recover(THD *thd)
if (protocol->send_fields(&field_list, if (protocol->send_fields(&field_list,
Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF)) Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
DBUG_RETURN(TRUE);
for (len= MAX_XID_LIST_SIZE ; list==0 && len > MIN_XID_LIST_SIZE; len/=2)
{
list=(XID *)my_malloc(len*sizeof(XID), MYF(0));
}
if (!list)
{
my_error(ER_OUTOFMEMORY, MYF(0), len);
DBUG_RETURN(1); DBUG_RETURN(1);
}
for ( ; ht < end_ht ; ht++) pthread_mutex_lock(&LOCK_xid_cache);
while (xs=(XID_STATE*)hash_element(&xid_cache, i++))
{ {
if (!(*ht)->recover) if (xs->xa_state==XA_PREPARED)
continue;
while ((got=(*(*ht)->recover)(list, len)) > 0 )
{ {
XID *xid, *end;
for (xid=list, end=list+got; xid < end; xid++)
{
if (xid->get_my_xid())
continue; // skip "our" xids
protocol->prepare_for_resend(); protocol->prepare_for_resend();
protocol->store_longlong((longlong)xid->formatID, FALSE); protocol->store_longlong((longlong)xs->xid.formatID, FALSE);
protocol->store_longlong((longlong)xid->gtrid_length, FALSE); protocol->store_longlong((longlong)xs->xid.gtrid_length, FALSE);
protocol->store_longlong((longlong)xid->bqual_length, FALSE); protocol->store_longlong((longlong)xs->xid.bqual_length, FALSE);
protocol->store(xid->data, xid->gtrid_length+xid->bqual_length, protocol->store(xs->xid.data, xs->xid.gtrid_length+xs->xid.bqual_length,
&my_charset_bin); &my_charset_bin);
if (protocol->write()) if (protocol->write())
goto err; {
pthread_mutex_unlock(&LOCK_xid_cache);
DBUG_RETURN(1);
} }
if (got < len)
break;
} }
} }
error=FALSE; pthread_mutex_unlock(&LOCK_xid_cache);
send_eof(thd); send_eof(thd);
err: DBUG_RETURN(0);
my_free((gptr)list, MYF(0));
DBUG_RETURN(error);
} }
/* /*
......
sql/handler.h
View file @ a0725e13
...@@ -227,11 +227,11 @@ struct xid_t { ...@@ -227,11 +227,11 @@ struct xid_t {
char data[XIDDATASIZE]; // not \0-terminated ! char data[XIDDATASIZE]; // not \0-terminated !
bool eq(struct xid_t *xid) bool eq(struct xid_t *xid)
{ return !memcmp(this, xid, sizeof(long)*3+gtrid_length+bqual_length); } { return !memcmp(this, xid, length()); }
bool eq(long g, long b, const char *d) bool eq(long g, long b, const char *d)
{ return g == gtrid_length && b == bqual_length && !memcmp(d, data, g+b); } { return g == gtrid_length && b == bqual_length && !memcmp(d, data, g+b); }
void set(struct xid_t *xid) void set(struct xid_t *xid)
{ memcpy(this, xid, sizeof(long)*3+xid->gtrid_length+xid->bqual_length); } { memcpy(this, xid, xid->length()); }
void set(long f, const char *g, long gl, const char *b, long bl) void set(long f, const char *g, long gl, const char *b, long bl)
{ {
formatID= f; formatID= f;
...@@ -270,6 +270,11 @@ struct xid_t { ...@@ -270,6 +270,11 @@ struct xid_t {
!memcmp(data, MYSQL_XID_PREFIX, MYSQL_XID_PREFIX_LEN) ? !memcmp(data, MYSQL_XID_PREFIX, MYSQL_XID_PREFIX_LEN) ?
quick_get_my_xid() : 0; quick_get_my_xid() : 0;
} }
uint length()
{
return sizeof(formatID)+sizeof(gtrid_length)+sizeof(bqual_length)+
gtrid_length+bqual_length;
}
}; };
typedef struct xid_t XID; typedef struct xid_t XID;
......
sql/mysqld.cc
View file @ a0725e13
...@@ -1050,6 +1050,7 @@ void clean_up(bool print_message) ...@@ -1050,6 +1050,7 @@ void clean_up(bool print_message)
(void) ha_panic(HA_PANIC_CLOSE); /* close all tables and logs */ (void) ha_panic(HA_PANIC_CLOSE); /* close all tables and logs */
if (tc_log) if (tc_log)
tc_log->close(); tc_log->close();
xid_cache_free();
delete_elements(&key_caches, (void (*)(const char*, gptr)) free_key_cache); delete_elements(&key_caches, (void (*)(const char*, gptr)) free_key_cache);
multi_keycache_free(); multi_keycache_free();
end_thr_alarm(1); /* Free allocated memory */ end_thr_alarm(1); /* Free allocated memory */
...@@ -2920,6 +2921,11 @@ server."); ...@@ -2920,6 +2921,11 @@ server.");
using_update_log=1; using_update_log=1;
} }
if (xid_cache_init())
{
sql_print_error("Out of memory");
unireg_abort(1);
}
if (ha_init()) if (ha_init())
{ {
sql_print_error("Can't init databases"); sql_print_error("Can't init databases");
......
sql/share/errmsg.txt
View file @ a0725e13
...@@ -5388,3 +5388,5 @@ ER_STACK_OVERRUN_NEED_MORE ...@@ -5388,3 +5388,5 @@ ER_STACK_OVERRUN_NEED_MORE
eng "Thread stack overrun: %ld bytes used of a %ld byte stack, and %ld bytes needed. Use 'mysqld -O thread_stack=#' to specify a bigger stack." eng "Thread stack overrun: %ld bytes used of a %ld byte stack, and %ld bytes needed. Use 'mysqld -O thread_stack=#' to specify a bigger stack."
ER_TOO_LONG_BODY 42000 S1009 ER_TOO_LONG_BODY 42000 S1009
eng "Routine body for '%-.100s' is too long" eng "Routine body for '%-.100s' is too long"
ER_XAER_DUPID XAE08
eng "XAER_DUPID: The XID already exists"
sql/sql_base.cc
View file @ a0725e13
...@@ -502,7 +502,7 @@ void close_thread_tables(THD *thd, bool lock_in_use, bool skip_derived, ...@@ -502,7 +502,7 @@ void close_thread_tables(THD *thd, bool lock_in_use, bool skip_derived,
*/ */
bzero(&thd->transaction.stmt, sizeof(thd->transaction.stmt)); bzero(&thd->transaction.stmt, sizeof(thd->transaction.stmt));
if (!thd->active_transaction()) if (!thd->active_transaction())
thd->transaction.xid.null(); thd->transaction.xid_state.xid.null();
/* VOID(pthread_sigmask(SIG_SETMASK,&thd->block_signals,NULL)); */ /* VOID(pthread_sigmask(SIG_SETMASK,&thd->block_signals,NULL)); */
if (!lock_in_use) if (!lock_in_use)
......
sql/sql_class.cc
View file @ a0725e13
...@@ -323,7 +323,8 @@ void THD::init_for_queries() ...@@ -323,7 +323,8 @@ void THD::init_for_queries()
variables.trans_alloc_block_size, variables.trans_alloc_block_size,
variables.trans_prealloc_size); variables.trans_prealloc_size);
#endif #endif
transaction.xid.null(); transaction.xid_state.xid.null();
transaction.xid_state.in_thd=1;
} }
...@@ -358,9 +359,15 @@ void THD::cleanup(void) ...@@ -358,9 +359,15 @@ void THD::cleanup(void)
{ {
DBUG_ENTER("THD::cleanup"); DBUG_ENTER("THD::cleanup");
#ifdef ENABLE_WHEN_BINLOG_WILL_BE_ABLE_TO_PREPARE #ifdef ENABLE_WHEN_BINLOG_WILL_BE_ABLE_TO_PREPARE
if (transaction.xa_state != XA_PREPARED) if (transaction.xid_state.xa_state == XA_PREPARED)
{
#error xid_state in the cache should be replaced by the allocated value
}
#endif #endif
{
ha_rollback(this); ha_rollback(this);
xid_cache_delete(&transaction.xid_state);
}
if (locked_tables) if (locked_tables)
{ {
lock=locked_tables; locked_tables=0; lock=locked_tables; locked_tables=0;
...@@ -1841,3 +1848,81 @@ void THD::pop_open_tables_state() ...@@ -1841,3 +1848,81 @@ void THD::pop_open_tables_state()
set_open_tables_state(state); set_open_tables_state(state);
DBUG_VOID_RETURN; DBUG_VOID_RETURN;
} }
pthread_mutex_t LOCK_xid_cache;
HASH xid_cache;
static byte *xid_get_hash_key(const byte *ptr,uint *length,
my_bool not_used __attribute__((unused)))
{
*length=((XID_STATE*)ptr)->xid.length();
return (byte *)&((XID_STATE*)ptr)->xid;
}
static void xid_free_hash (void *ptr)
{
if (!((XID_STATE*)ptr)->in_thd)
my_free((byte *)ptr, MYF(0));
}
bool xid_cache_init()
{
pthread_mutex_init(&LOCK_xid_cache, MY_MUTEX_INIT_FAST);
hash_init(&xid_cache, &my_charset_bin, 100, 0, 0,
xid_get_hash_key, xid_free_hash, 0) != 0;
}
void xid_cache_free()
{
if (hash_inited(&xid_cache))
{
hash_free(&xid_cache);
pthread_mutex_destroy(&LOCK_xid_cache);
}
}
XID_STATE *xid_cache_search(XID *xid)
{
pthread_mutex_lock(&LOCK_xid_cache);
XID_STATE *res=(XID_STATE *)hash_search(&xid_cache, (byte *)xid, xid->length());
pthread_mutex_unlock(&LOCK_xid_cache);
return res;
}
bool xid_cache_insert(XID *xid, enum xa_states xa_state)
{
XID_STATE *xs;
my_bool res;
pthread_mutex_lock(&LOCK_xid_cache);
if (hash_search(&xid_cache, (byte *)xid, xid->length()))
res=0;
else if (!(xs=(XID_STATE *)my_malloc(sizeof(*xs), MYF(MY_WME))))
res=1;
else
{
xs->xa_state=xa_state;
xs->xid.set(xid);
xs->in_thd=0;
res=my_hash_insert(&xid_cache, (byte*)xs);
}
pthread_mutex_unlock(&LOCK_xid_cache);
return res;
}
bool xid_cache_insert(XID_STATE *xid_state)
{
pthread_mutex_lock(&LOCK_xid_cache);
DBUG_ASSERT(hash_search(&xid_cache, (byte *)&xid_state->xid,
xid_state->xid.length())==0);
my_bool res=my_hash_insert(&xid_cache, (byte*)xid_state);
pthread_mutex_unlock(&LOCK_xid_cache);
return res;
}
void xid_cache_delete(XID_STATE *xid_state)
{
pthread_mutex_lock(&LOCK_xid_cache);
hash_delete(&xid_cache, (byte *)xid_state);
pthread_mutex_unlock(&LOCK_xid_cache);
}
sql/sql_class.h
View file @ a0725e13
...@@ -351,8 +351,6 @@ class MYSQL_LOG: public TC_LOG ...@@ -351,8 +351,6 @@ class MYSQL_LOG: public TC_LOG
inline uint32 get_open_count() { return open_count; } inline uint32 get_open_count() { return open_count; }
}; };
/* character conversion tables */
typedef struct st_copy_info { typedef struct st_copy_info {
ha_rows records; ha_rows records;
...@@ -631,7 +629,7 @@ typedef struct system_status_var ...@@ -631,7 +629,7 @@ typedef struct system_status_var
ulong filesort_range_count; ulong filesort_range_count;
ulong filesort_rows; ulong filesort_rows;
ulong filesort_scan_count; ulong filesort_scan_count;
/* Ppepared statements and binary protocol */ /* Prepared statements and binary protocol */
ulong com_stmt_prepare; ulong com_stmt_prepare;
ulong com_stmt_execute; ulong com_stmt_execute;
ulong com_stmt_send_long_data; ulong com_stmt_send_long_data;
...@@ -925,6 +923,22 @@ struct st_savepoint { ...@@ -925,6 +923,22 @@ struct st_savepoint {
enum xa_states {XA_NOTR=0, XA_ACTIVE, XA_IDLE, XA_PREPARED}; enum xa_states {XA_NOTR=0, XA_ACTIVE, XA_IDLE, XA_PREPARED};
extern const char *xa_state_names[]; extern const char *xa_state_names[];
typedef struct st_xid_state {
/* For now, this is only used to catch duplicated external xids */
XID xid; // transaction identifier
enum xa_states xa_state; // used by external XA only
bool in_thd;
} XID_STATE;
extern pthread_mutex_t LOCK_xid_cache;
extern HASH xid_cache;
bool xid_cache_init(void);
void xid_cache_free(void);
XID_STATE *xid_cache_search(XID *xid);
bool xid_cache_insert(XID *xid, enum xa_states xa_state);
bool xid_cache_insert(XID_STATE *xid_state);
void xid_cache_delete(XID_STATE *xid_state);
/* /*
A registry for item tree transformations performed during A registry for item tree transformations performed during
query optimization. We register only those changes which require query optimization. We register only those changes which require
...@@ -946,7 +960,7 @@ enum prelocked_mode_type {NON_PRELOCKED= 0, PRELOCKED= 1, ...@@ -946,7 +960,7 @@ enum prelocked_mode_type {NON_PRELOCKED= 0, PRELOCKED= 1,
/* /*
Class that holds information about tables which were open and locked Class that holds information about tables which were opened and locked
by the thread. It is also used to save/restore this information in by the thread. It is also used to save/restore this information in
push_open_tables_state()/pop_open_tables_state(). push_open_tables_state()/pop_open_tables_state().
*/ */
...@@ -1143,8 +1157,7 @@ class THD :public Statement, ...@@ -1143,8 +1157,7 @@ class THD :public Statement,
THD_TRANS all; // Trans since BEGIN WORK THD_TRANS all; // Trans since BEGIN WORK
THD_TRANS stmt; // Trans for current statement THD_TRANS stmt; // Trans for current statement
bool on; // see ha_enable_transaction() bool on; // see ha_enable_transaction()
XID xid; // transaction identifier XID_STATE xid_state;
enum xa_states xa_state; // used by external XA only
/* /*
Tables changed in transaction (that must be invalidated in query cache). Tables changed in transaction (that must be invalidated in query cache).
List contain only transactional tables, that not invalidated in query List contain only transactional tables, that not invalidated in query
...@@ -1164,7 +1177,7 @@ class THD :public Statement, ...@@ -1164,7 +1177,7 @@ class THD :public Statement,
st_transactions() st_transactions()
{ {
bzero((char*)this, sizeof(*this)); bzero((char*)this, sizeof(*this));
xid.null(); xid_state.xid.null();
init_sql_alloc(&mem_root, ALLOC_ROOT_MIN_BLOCK_SIZE, 0); init_sql_alloc(&mem_root, ALLOC_ROOT_MIN_BLOCK_SIZE, 0);
} }
#endif #endif
......
sql/sql_parse.cc
View file @ a0725e13
...@@ -2017,7 +2017,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd, ...@@ -2017,7 +2017,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
*/ */
bzero(&thd->transaction.stmt, sizeof(thd->transaction.stmt)); bzero(&thd->transaction.stmt, sizeof(thd->transaction.stmt));
if (!thd->active_transaction()) if (!thd->active_transaction())
thd->transaction.xid.null(); thd->transaction.xid_state.xid.null();
/* report error issued during command execution */ /* report error issued during command execution */
if (thd->killed_errno() && !thd->net.report_error) if (thd->killed_errno() && !thd->net.report_error)
...@@ -4502,14 +4502,15 @@ mysql_execute_command(THD *thd) ...@@ -4502,14 +4502,15 @@ mysql_execute_command(THD *thd)
break; break;
} }
case SQLCOM_XA_START: case SQLCOM_XA_START:
if (thd->transaction.xa_state == XA_IDLE && thd->lex->xa_opt == XA_RESUME) if (thd->transaction.xid_state.xa_state == XA_IDLE &&
thd->lex->xa_opt == XA_RESUME)
{ {
if (! thd->transaction.xid.eq(thd->lex->xid)) if (! thd->transaction.xid_state.xid.eq(thd->lex->xid))
{ {
my_error(ER_XAER_NOTA, MYF(0)); my_error(ER_XAER_NOTA, MYF(0));
break; break;
} }
thd->transaction.xa_state=XA_ACTIVE; thd->transaction.xid_state.xa_state=XA_ACTIVE;
send_ok(thd); send_ok(thd);
break; break;
} }
...@@ -4518,10 +4519,10 @@ mysql_execute_command(THD *thd) ...@@ -4518,10 +4519,10 @@ mysql_execute_command(THD *thd)
my_error(ER_XAER_INVAL, MYF(0)); my_error(ER_XAER_INVAL, MYF(0));
break; break;
} }
if (thd->transaction.xa_state != XA_NOTR) if (thd->transaction.xid_state.xa_state != XA_NOTR)
{ {
my_error(ER_XAER_RMFAIL, MYF(0), my_error(ER_XAER_RMFAIL, MYF(0),
xa_state_names[thd->transaction.xa_state]); xa_state_names[thd->transaction.xid_state.xa_state]);
break; break;
} }
if (thd->active_transaction() || thd->locked_tables) if (thd->active_transaction() || thd->locked_tables)
...@@ -4529,9 +4530,15 @@ mysql_execute_command(THD *thd) ...@@ -4529,9 +4530,15 @@ mysql_execute_command(THD *thd)
my_error(ER_XAER_OUTSIDE, MYF(0)); my_error(ER_XAER_OUTSIDE, MYF(0));
break; break;
} }
DBUG_ASSERT(thd->transaction.xid.is_null()); if (xid_cache_search(thd->lex->xid))
thd->transaction.xa_state=XA_ACTIVE; {
thd->transaction.xid.set(thd->lex->xid); my_error(ER_XAER_DUPID, MYF(0));
break;
}
DBUG_ASSERT(thd->transaction.xid_state.xid.is_null());
thd->transaction.xid_state.xa_state=XA_ACTIVE;
thd->transaction.xid_state.xid.set(thd->lex->xid);
xid_cache_insert(&thd->transaction.xid_state);
thd->options= ((thd->options & (ulong) ~(OPTION_STATUS_NO_TRANS_UPDATE)) | thd->options= ((thd->options & (ulong) ~(OPTION_STATUS_NO_TRANS_UPDATE)) |
OPTION_BEGIN); OPTION_BEGIN);
thd->server_status|= SERVER_STATUS_IN_TRANS; thd->server_status|= SERVER_STATUS_IN_TRANS;
...@@ -4544,28 +4551,28 @@ mysql_execute_command(THD *thd) ...@@ -4544,28 +4551,28 @@ mysql_execute_command(THD *thd)
my_error(ER_XAER_INVAL, MYF(0)); my_error(ER_XAER_INVAL, MYF(0));
break; break;
} }
if (thd->transaction.xa_state != XA_ACTIVE) if (thd->transaction.xid_state.xa_state != XA_ACTIVE)
{ {
my_error(ER_XAER_RMFAIL, MYF(0), my_error(ER_XAER_RMFAIL, MYF(0),
xa_state_names[thd->transaction.xa_state]); xa_state_names[thd->transaction.xid_state.xa_state]);
break; break;
} }
if (!thd->transaction.xid.eq(thd->lex->xid)) if (!thd->transaction.xid_state.xid.eq(thd->lex->xid))
{ {
my_error(ER_XAER_NOTA, MYF(0)); my_error(ER_XAER_NOTA, MYF(0));
break; break;
} }
thd->transaction.xa_state=XA_IDLE; thd->transaction.xid_state.xa_state=XA_IDLE;
send_ok(thd); send_ok(thd);
break; break;
case SQLCOM_XA_PREPARE: case SQLCOM_XA_PREPARE:
if (thd->transaction.xa_state != XA_IDLE) if (thd->transaction.xid_state.xa_state != XA_IDLE)
{ {
my_error(ER_XAER_RMFAIL, MYF(0), my_error(ER_XAER_RMFAIL, MYF(0),
xa_state_names[thd->transaction.xa_state]); xa_state_names[thd->transaction.xid_state.xa_state]);
break; break;
} }
if (!thd->transaction.xid.eq(thd->lex->xid)) if (!thd->transaction.xid_state.xid.eq(thd->lex->xid))
{ {
my_error(ER_XAER_NOTA, MYF(0)); my_error(ER_XAER_NOTA, MYF(0));
break; break;
...@@ -4573,22 +4580,28 @@ mysql_execute_command(THD *thd) ...@@ -4573,22 +4580,28 @@ mysql_execute_command(THD *thd)
if (ha_prepare(thd)) if (ha_prepare(thd))
{ {
my_error(ER_XA_RBROLLBACK, MYF(0)); my_error(ER_XA_RBROLLBACK, MYF(0));
thd->transaction.xa_state=XA_NOTR; xid_cache_delete(&thd->transaction.xid_state);
thd->transaction.xid_state.xa_state=XA_NOTR;
break; break;
} }
thd->transaction.xa_state=XA_PREPARED; thd->transaction.xid_state.xa_state=XA_PREPARED;
send_ok(thd); send_ok(thd);
break; break;
case SQLCOM_XA_COMMIT: case SQLCOM_XA_COMMIT:
if (!thd->transaction.xid.eq(thd->lex->xid)) if (!thd->transaction.xid_state.xid.eq(thd->lex->xid))
{ {
if (!(res= !ha_commit_or_rollback_by_xid(thd->lex->xid, 1))) XID_STATE *xs=xid_cache_search(thd->lex->xid);
if (!xs || xs->in_thd)
my_error(ER_XAER_NOTA, MYF(0)); my_error(ER_XAER_NOTA, MYF(0));
else else
{
ha_commit_or_rollback_by_xid(thd->lex->xid, 1);
xid_cache_delete(xs);
send_ok(thd); send_ok(thd);
}
break; break;
} }
if (thd->transaction.xa_state == XA_IDLE && if (thd->transaction.xid_state.xa_state == XA_IDLE &&
thd->lex->xa_opt == XA_ONE_PHASE) thd->lex->xa_opt == XA_ONE_PHASE)
{ {
int r; int r;
...@@ -4597,7 +4610,7 @@ mysql_execute_command(THD *thd) ...@@ -4597,7 +4610,7 @@ mysql_execute_command(THD *thd)
else else
send_ok(thd); send_ok(thd);
} }
else if (thd->transaction.xa_state == XA_PREPARED && else if (thd->transaction.xid_state.xa_state == XA_PREPARED &&
thd->lex->xa_opt == XA_NONE) thd->lex->xa_opt == XA_NONE)
{ {
if (wait_if_global_read_lock(thd, 0, 0)) if (wait_if_global_read_lock(thd, 0, 0))
...@@ -4617,27 +4630,33 @@ mysql_execute_command(THD *thd) ...@@ -4617,27 +4630,33 @@ mysql_execute_command(THD *thd)
else else
{ {
my_error(ER_XAER_RMFAIL, MYF(0), my_error(ER_XAER_RMFAIL, MYF(0),
xa_state_names[thd->transaction.xa_state]); xa_state_names[thd->transaction.xid_state.xa_state]);
break; break;
} }
thd->options&= ~(ulong) (OPTION_BEGIN | OPTION_STATUS_NO_TRANS_UPDATE); thd->options&= ~(ulong) (OPTION_BEGIN | OPTION_STATUS_NO_TRANS_UPDATE);
thd->server_status&= ~SERVER_STATUS_IN_TRANS; thd->server_status&= ~SERVER_STATUS_IN_TRANS;
thd->transaction.xa_state=XA_NOTR; xid_cache_delete(&thd->transaction.xid_state);
thd->transaction.xid_state.xa_state=XA_NOTR;
break; break;
case SQLCOM_XA_ROLLBACK: case SQLCOM_XA_ROLLBACK:
if (!thd->transaction.xid.eq(thd->lex->xid)) if (!thd->transaction.xid_state.xid.eq(thd->lex->xid))
{ {
if (!(res= !ha_commit_or_rollback_by_xid(thd->lex->xid, 0))) XID_STATE *xs=xid_cache_search(thd->lex->xid);
if (!xs || xs->in_thd)
my_error(ER_XAER_NOTA, MYF(0)); my_error(ER_XAER_NOTA, MYF(0));
else else
{
ha_commit_or_rollback_by_xid(thd->lex->xid, 0);
xid_cache_delete(xs);
send_ok(thd); send_ok(thd);
}
break; break;
} }
if (thd->transaction.xa_state != XA_IDLE && if (thd->transaction.xid_state.xa_state != XA_IDLE &&
thd->transaction.xa_state != XA_PREPARED) thd->transaction.xid_state.xa_state != XA_PREPARED)
{ {
my_error(ER_XAER_RMFAIL, MYF(0), my_error(ER_XAER_RMFAIL, MYF(0),
xa_state_names[thd->transaction.xa_state]); xa_state_names[thd->transaction.xid_state.xa_state]);
break; break;
} }
if (ha_rollback(thd)) if (ha_rollback(thd))
...@@ -4646,7 +4665,8 @@ mysql_execute_command(THD *thd) ...@@ -4646,7 +4665,8 @@ mysql_execute_command(THD *thd)
send_ok(thd); send_ok(thd);
thd->options&= ~(ulong) (OPTION_BEGIN | OPTION_STATUS_NO_TRANS_UPDATE); thd->options&= ~(ulong) (OPTION_BEGIN | OPTION_STATUS_NO_TRANS_UPDATE);
thd->server_status&= ~SERVER_STATUS_IN_TRANS; thd->server_status&= ~SERVER_STATUS_IN_TRANS;
thd->transaction.xa_state=XA_NOTR; xid_cache_delete(&thd->transaction.xid_state);
thd->transaction.xid_state.xa_state=XA_NOTR;
break; break;
case SQLCOM_XA_RECOVER: case SQLCOM_XA_RECOVER:
res= mysql_xa_recover(thd); res= mysql_xa_recover(thd);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7