database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-07-30 16:24:05 +03:00
Code Activity

ndb - bug#10029 fix

Browse Source 
ndb/include/kernel/signaldata/TuxMaint.hpp:
  handle multipl index update failure atomically
ndb/src/kernel/blocks/dbtup/Dbtup.hpp:
  handle multipl index update failure atomically
ndb/src/kernel/blocks/dbtup/DbtupTrigger.cpp:
  handle multipl index update failure atomically
ndb/src/kernel/blocks/dbtup/Notes.txt:
  handle multipl index update failure atomically
ndb/src/kernel/blocks/dbtux/DbtuxNode.cpp:
  handle multipl index update failure atomically
ndb/src/ndbapi/ndberror.c:
  handle multipl index update failure atomically
ndb/test/ndbapi/testOIBasic.cpp:
  handle multipl index update failure atomically
This commit is contained in:
unknown
2005-04-23 17:33:12 +02:00
parent af4c58e090
commit ffb64880b7
7 changed files with 205 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

160
ndb/test/ndbapi/testOIBasic.cpp
View File

@ -228,6 +228,8 @@ struct Par : public Opt {
bool m_verify;
// deadlock possible
bool m_deadlock;
// abort percentabge
unsigned m_abortpct;
// timer location
Par(const Opt& opt) :
Opt(opt),
@ -243,7 +245,8 @@ struct Par : public Opt {
m_pctrange(0),
m_randomkey(false),
m_verify(false),
m_deadlock(false) {
m_deadlock(false),
m_abortpct(0) {
}
};
@ -684,7 +687,7 @@ struct Con {
NdbResultSet* m_resultset;
enum ScanMode { ScanNo = 0, Committed, Latest, Exclusive };
ScanMode m_scanmode;
enum ErrType { ErrNone = 0, ErrDeadlock, ErrOther };
enum ErrType { ErrNone = 0, ErrDeadlock, ErrNospace, ErrOther };
ErrType m_errtype;
Con() :
m_ndb(0), m_dic(0), m_tx(0), m_op(0),
@ -705,7 +708,7 @@ struct Con {
int setValue(int num, const char* addr);
int setBound(int num, int type, const void* value);
int execute(ExecType t);
int execute(ExecType t, bool& deadlock);
int execute(ExecType t, bool& deadlock, bool& nospace);
int openScanRead(unsigned scanbat, unsigned scanpar);
int openScanExclusive(unsigned scanbat, unsigned scanpar);
int executeScan();
@ -818,17 +821,21 @@ Con::execute(ExecType t)
}
int
Con::execute(ExecType t, bool& deadlock)
Con::execute(ExecType t, bool& deadlock, bool& nospace)
{
int ret = execute(t);
if (ret != 0) {
if (deadlock && m_errtype == ErrDeadlock) {
LL3("caught deadlock");
ret = 0;
}
if (ret != 0 && deadlock && m_errtype == ErrDeadlock) {
LL3("caught deadlock");
ret = 0;
} else {
deadlock = false;
}
if (ret != 0 && nospace && m_errtype == ErrNospace) {
LL3("caught nospace");
ret = 0;
} else {
nospace = false;
}
CHK(ret == 0);
return 0;
}
@ -940,6 +947,8 @@ Con::printerror(NdbOut& out)
die += (code == g_opt.m_die);
if (code == 266 || code == 274 || code == 296 || code == 297 || code == 499)
m_errtype = ErrDeadlock;
if (code == 826 || code == 827 || code == 902)
m_errtype = ErrNospace;
}
if (m_op && m_op->getNdbError().code != 0) {
LL0(++any << " op : error " << m_op->getNdbError());
@ -1128,6 +1137,16 @@ irandom(unsigned n)
return i;
}
static bool
randompct(unsigned pct)
{
if (pct == 0)
return false;
if (pct >= 100)
return true;
return urandom(100) < pct;
}
// Val - typed column value
struct Val {
@ -1565,8 +1584,8 @@ struct Set {
// row methods
bool exist(unsigned i) const;
Row::Op pending(unsigned i) const;
void notpending(unsigned i);
void notpending(const Lst& lst);
void notpending(unsigned i, ExecType et = Commit);
void notpending(const Lst& lst, ExecType et = Commit);
void calc(Par par, unsigned i);
int insrow(Par par, unsigned i);
int updrow(Par par, unsigned i);
@ -1775,23 +1794,30 @@ Set::putval(unsigned i, bool force)
}
void
Set::notpending(unsigned i)
Set::notpending(unsigned i, ExecType et)
{
assert(m_row[i] != 0);
Row& row = *m_row[i];
if (row.m_pending == Row::InsOp)
row.m_exist = true;
if (row.m_pending == Row::DelOp)
row.m_exist = false;
if (et == Commit) {
if (row.m_pending == Row::InsOp)
row.m_exist = true;
if (row.m_pending == Row::DelOp)
row.m_exist = false;
} else {
if (row.m_pending == Row::InsOp)
row.m_exist = false;
if (row.m_pending == Row::DelOp)
row.m_exist = true;
}
row.m_pending = Row::NoOp;
}
void
Set::notpending(const Lst& lst)
Set::notpending(const Lst& lst, ExecType et)
{
for (unsigned j = 0; j < lst.m_cnt; j++) {
unsigned i = lst.m_arr[j];
notpending(i);
notpending(i, et);
}
}
@ -2121,14 +2147,20 @@ pkinsert(Par par)
lst.push(i);
if (lst.cnt() == par.m_batch) {
bool deadlock = par.m_deadlock;
CHK(con.execute(Commit, deadlock) == 0);
bool nospace = true;
ExecType et = randompct(par.m_abortpct) ? Rollback : Commit;
CHK(con.execute(et, deadlock, nospace) == 0);
con.closeTransaction();
if (deadlock) {
LL1("pkinsert: stop on deadlock");
return 0;
}
if (nospace) {
LL1("pkinsert: cnt=" << j << " stop on nospace");
return 0;
}
set.lock();
set.notpending(lst);
set.notpending(lst, et);
set.unlock();
lst.reset();
CHK(con.startTransaction() == 0);
@ -2136,14 +2168,20 @@ pkinsert(Par par)
}
if (lst.cnt() != 0) {
bool deadlock = par.m_deadlock;
CHK(con.execute(Commit, deadlock) == 0);
bool nospace = true;
ExecType et = randompct(par.m_abortpct) ? Rollback : Commit;
CHK(con.execute(et, deadlock, nospace) == 0);
con.closeTransaction();
if (deadlock) {
LL1("pkinsert: stop on deadlock");
return 0;
}
if (nospace) {
LL1("pkinsert: end: stop on nospace");
return 0;
}
set.lock();
set.notpending(lst);
set.notpending(lst, et);
set.unlock();
return 0;
}
@ -2160,6 +2198,7 @@ pkupdate(Par par)
CHK(con.startTransaction() == 0);
Lst lst;
bool deadlock = false;
bool nospace = false;
for (unsigned j = 0; j < par.m_rows; j++) {
unsigned j2 = ! par.m_randomkey ? j : urandom(par.m_rows);
unsigned i = thrrow(par, j2);
@ -2175,27 +2214,37 @@ pkupdate(Par par)
lst.push(i);
if (lst.cnt() == par.m_batch) {
deadlock = par.m_deadlock;
CHK(con.execute(Commit, deadlock) == 0);
nospace = true;
ExecType et = randompct(par.m_abortpct) ? Rollback : Commit;
CHK(con.execute(et, deadlock, nospace) == 0);
if (deadlock) {
LL1("pkupdate: stop on deadlock");
break;
}
if (nospace) {
LL1("pkupdate: cnt=" << j << " stop on nospace");
break;
}
con.closeTransaction();
set.lock();
set.notpending(lst);
set.notpending(lst, et);
set.unlock();
lst.reset();
CHK(con.startTransaction() == 0);
}
}
if (! deadlock && lst.cnt() != 0) {
if (! deadlock && ! nospace && lst.cnt() != 0) {
deadlock = par.m_deadlock;
CHK(con.execute(Commit, deadlock) == 0);
nospace = true;
ExecType et = randompct(par.m_abortpct) ? Rollback : Commit;
CHK(con.execute(et, deadlock, nospace) == 0);
if (deadlock) {
LL1("pkupdate: stop on deadlock");
} else if (nospace) {
LL1("pkupdate: end: stop on nospace");
} else {
set.lock();
set.notpending(lst);
set.notpending(lst, et);
set.unlock();
}
}
@ -2212,6 +2261,7 @@ pkdelete(Par par)
CHK(con.startTransaction() == 0);
Lst lst;
bool deadlock = false;
bool nospace = false;
for (unsigned j = 0; j < par.m_rows; j++) {
unsigned j2 = ! par.m_randomkey ? j : urandom(par.m_rows);
unsigned i = thrrow(par, j2);
@ -2226,27 +2276,31 @@ pkdelete(Par par)
lst.push(i);
if (lst.cnt() == par.m_batch) {
deadlock = par.m_deadlock;
CHK(con.execute(Commit, deadlock) == 0);
nospace = true;
ExecType et = randompct(par.m_abortpct) ? Rollback : Commit;
CHK(con.execute(et, deadlock, nospace) == 0);
if (deadlock) {
LL1("pkdelete: stop on deadlock");
break;
}
con.closeTransaction();
set.lock();
set.notpending(lst);
set.notpending(lst, et);
set.unlock();
lst.reset();
CHK(con.startTransaction() == 0);
}
}
if (! deadlock && lst.cnt() != 0) {
if (! deadlock && ! nospace && lst.cnt() != 0) {
deadlock = par.m_deadlock;
CHK(con.execute(Commit, deadlock) == 0);
nospace = true;
ExecType et = randompct(par.m_abortpct) ? Rollback : Commit;
CHK(con.execute(et, deadlock, nospace) == 0);
if (deadlock) {
LL1("pkdelete: stop on deadlock");
} else {
set.lock();
set.notpending(lst);
set.notpending(lst, et);
set.unlock();
}
}
@ -2730,6 +2784,10 @@ readverify(Par par)
if (par.m_noverify)
return 0;
par.m_verify = true;
if (par.m_abortpct != 0) {
LL2("skip verify in this version"); // implement in 5.0 version
par.m_verify = false;
}
CHK(pkread(par) == 0);
CHK(scanreadall(par) == 0);
return 0;
@ -3028,11 +3086,11 @@ runstep(Par par, const char* fname, TFunc func, unsigned mode)
for (n = 0; n < threads; n++) {
LL4("start " << n);
Thr& thr = *g_thrlist[n];
thr.m_par.m_tab = par.m_tab;
thr.m_par.m_set = par.m_set;
thr.m_par.m_tmr = par.m_tmr;
thr.m_par.m_lno = par.m_lno;
thr.m_par.m_slno = par.m_slno;
Par oldpar = thr.m_par;
// update parameters
thr.m_par = par;
thr.m_par.m_no = oldpar.m_no;
thr.m_par.m_con = oldpar.m_con;
thr.m_func = func;
thr.start();
}
@ -3143,6 +3201,24 @@ tbusybuild(Par par)
return 0;
}
static int
trollback(Par par)
{
par.m_abortpct = 50;
RUNSTEP(par, droptable, ST);
RUNSTEP(par, createtable, ST);
RUNSTEP(par, invalidatetable, MT);
RUNSTEP(par, pkinsert, MT);
RUNSTEP(par, createindex, ST);
RUNSTEP(par, invalidateindex, MT);
RUNSTEP(par, readverify, ST);
for (par.m_slno = 0; par.m_slno < par.m_subloop; par.m_slno++) {
RUNSTEP(par, mixedoperations, MT);
RUNSTEP(par, readverify, ST);
}
return 0;
}
static int
ttimebuild(Par par)
{
@ -3252,10 +3328,12 @@ struct TCase {
static const TCase
tcaselist[] = {
TCase("a", tbuild, "index build"),
TCase("b", tpkops, "pk operations"),
TCase("c", tpkopsread, "pk operations and scan reads"),
TCase("d", tmixedops, "pk operations and scan operations"),
TCase("e", tbusybuild, "pk operations and index build"),
// "b" in 5.0
TCase("c", tpkops, "pk operations"),
TCase("d", tpkopsread, "pk operations and scan reads"),
TCase("e", tmixedops, "pk operations and scan operations"),
TCase("f", tbusybuild, "pk operations and index build"),
TCase("g", trollback, "operations with random rollbacks"),
TCase("t", ttimebuild, "time index build"),
TCase("u", ttimemaint, "time index maintenance"),
TCase("v", ttimescan, "time full scan table vs index on pk"),