4
#include <NdbRestarter.hpp>
5
#include <HugoOperations.hpp>
6
#include <HugoTransactions.hpp>
7
#include <UtilTransactions.hpp>
8
#include <signaldata/DumpStateOrd.hpp>
11
#include <InputStream.hpp>
24
static CASE g_op_types[] =
26
{ false, true, false, "INS", 0, 0, 0 }, // 0x001 a
27
{ true, true, false, "UPD", 0, 0, 0 }, // 0x002 d
28
{ true, false, false, "DEL", 0, 0, 0 }, // 0x004 g
30
{ false, true, false, "INS", "UPD", 0, 0 }, // 0x008 b
31
{ false, false, false, "INS", "DEL", 0, 0 }, // 0x010 c
32
{ true, true, false, "UPD", "UPD", 0, 0 }, // 0x020 e
33
{ true, false, false, "UPD", "DEL", 0, 0 }, // 0x040 f
34
{ true, true, false, "DEL", "INS", 0, 0 }, // 0x080 h
36
{ false, true, false, "INS", "DEL", "INS", 0 }, // 0x100 i
37
{ true, false, false, "DEL", "INS", "DEL", 0 } // 0x200 j
39
const size_t OP_COUNT = (sizeof(g_op_types)/sizeof(g_op_types[0]));
41
static Ndb* g_ndb = 0;
43
static Ndb_cluster_connection *g_cluster_connection= 0;
44
static HugoOperations* g_hugo_ops;
45
static int g_use_ops = 1 | 2 | 4;
46
static int g_cases = 0x1;
47
static int g_case_loop = 2;
48
static int g_rows = 10;
49
static int g_setup_tables = 1;
50
static int g_one_op_at_a_time = 0;
51
static const char * g_tablename = "T1";
52
static const NdbDictionary::Table* g_table = 0;
53
static NdbRestarter g_restarter;
55
static int init_ndb(int argc, char** argv);
56
static int parse_args(int argc, char** argv);
57
static int connect_ndb();
58
static int drop_all_tables();
59
static int load_table();
60
static int pause_lcp(int error);
61
static int do_op(int row);
62
static int continue_lcp(int error = 0);
65
static int validate();
67
#define require(x) { bool b = x; if(!b){g_err << __LINE__ << endl; abort();}}
70
main(int argc, char ** argv){
72
require(!init_ndb(argc, argv));
73
if(parse_args(argc, argv))
75
require(!connect_ndb());
78
require(!drop_all_tables());
80
if(NDBT_Tables::createTable(g_ndb, g_tablename) != 0){
85
g_table = g_ndb->getDictionary()->getTable(g_tablename);
87
g_err << "Failed to retreive table: " << g_tablename << endl;
90
require(g_hugo_ops = new HugoOperations(* g_table));
91
require(!g_hugo_ops->startTransaction(g_ndb));
93
g_ops= new CASE[g_rows];
95
const int use_ops = g_use_ops;
96
for(size_t i = 0; i<OP_COUNT; i++)
98
if(g_one_op_at_a_time){
99
while(i < OP_COUNT && (use_ops & (1 << i)) == 0) i++;
102
ndbout_c("-- loop\noperation: %c use_ops: %x", 'a'+i, use_ops);
103
g_use_ops = (1 << i);
108
size_t test_case = 0;
109
if((1 << test_case++) & g_cases)
111
for(size_t tl = 0; tl<g_case_loop; tl++){
112
g_info << "Performing all ops wo/ inteference of LCP" << endl;
114
g_info << "Testing pre LCP operations, ZLCP_OP_WRITE_RT_BREAK" << endl;
115
g_info << " where ZLCP_OP_WRITE_RT_BREAK is "
116
" finished before SAVE_PAGES" << endl;
117
require(!load_table());
118
require(!pause_lcp(5900));
119
for(size_t j = 0; j<g_rows; j++){
122
require(!continue_lcp(5900));
124
require(!pause_lcp(5900));
126
require(!validate());
130
if((1 << test_case++) & g_cases)
132
for(size_t tl = 0; tl<g_case_loop; tl++){
133
g_info << "Testing pre LCP operations, ZLCP_OP_WRITE_RT_BREAK" << endl;
134
g_info << " where ZLCP_OP_WRITE_RT_BREAK is finished after SAVE_PAGES"
136
require(!load_table());
137
require(!pause_lcp(5901));
138
for(size_t j = 0; j<g_rows; j++){
141
require(!continue_lcp(5901));
143
require(!pause_lcp(5900));
145
require(!validate());
149
if((1 << test_case++) & g_cases)
151
for(size_t tl = 0; tl<g_case_loop; tl++){
152
g_info << "Testing pre LCP operations, undo-ed at commit" << endl;
153
require(!load_table());
154
require(!pause_lcp(5902));
155
for(size_t j = 0; j<g_rows; j++){
158
require(!continue_lcp(5902));
160
require(!continue_lcp(5903));
161
require(!pause_lcp(5900));
163
require(!validate());
167
if((1 << test_case++) & g_cases)
169
for(size_t tl = 0; tl<g_case_loop; tl++){
170
g_info << "Testing prepared during LCP and committed after" << endl;
171
require(!load_table());
172
require(!pause_lcp(5904)); // Start LCP, but don't save pages
173
for(size_t j = 0; j<g_rows; j++){
176
require(!continue_lcp(5904)); // Start ACC save pages
177
require(!pause_lcp(5900)); // Next LCP
180
require(!validate());
186
static int init_ndb(int argc, char** argv)
192
static int parse_args(int argc, char** argv)
195
char * ops= 0, *cases=0;
196
struct getargs args[] = {
197
{ "records", 0, arg_integer, &g_rows, "Number of records", "records" },
198
{ "operations", 'o', arg_string, &ops, "Operations [a-h]", 0 },
199
{ "1", '1', arg_flag, &g_one_op_at_a_time, "One op at a time", 0 },
200
{ "0", '0', arg_negative_flag, &g_one_op_at_a_time, "All ops at once", 0 },
201
{ "cases", 'c', arg_string, &cases, "Cases [a-c]", 0 },
202
{ 0, 't', arg_flag, &g_setup_tables, "Create table", 0 },
203
{ 0, 'u', arg_negative_flag, &g_setup_tables, "Dont create table", 0 }
207
const int num_args = sizeof(args)/sizeof(args[0]);
208
if(getarg(args, num_args, argc, (const char**)argv, &optind)) {
209
arg_printusage(args, num_args, argv[0], " tabname1\n");
210
ndbout_c("\n -- Operations [a-%c] = ", 'a'+OP_COUNT-1);
211
for(i = 0; i<OP_COUNT; i++){
212
ndbout_c("\t%c = %s %s",
213
'a'+i, g_op_types[i].op1,
214
g_op_types[i].op2 ? g_op_types[i].op2 : "");
223
g_use_ops |= (1 << ((* s++) - 'a'));
230
g_cases |= (1 << ((* s++) - 'a'));
233
ndbout_c("table: %s", g_tablename);
234
printf("operations: ");
235
for(i = 0; i<OP_COUNT; i++)
236
if(g_use_ops & (1 << i))
240
printf("test cases: ");
242
if(g_cases & (1 << i))
245
printf("-------------\n");
249
static int connect_ndb()
251
g_cluster_connection = new Ndb_cluster_connection();
252
if(g_cluster_connection->connect(12, 5, 1) != 0)
257
g_ndb = new Ndb(g_cluster_connection, "TEST_DB");
259
if(g_ndb->waitUntilReady(30) == 0){
261
// int args[] = { DumpStateOrd::DihMaxTimeBetweenLCP };
262
// return g_restarter.dumpStateAllNodes(args, 1);
267
static int disconnect_ndb()
270
delete g_cluster_connection;
273
g_cluster_connection= 0;
277
static int drop_all_tables()
279
NdbDictionary::Dictionary * dict = g_ndb->getDictionary();
282
BaseString db = g_ndb->getDatabaseName();
283
BaseString schema = g_ndb->getSchemaName();
285
NdbDictionary::Dictionary::List list;
286
if (dict->listObjects(list, NdbDictionary::Object::TypeUndefined) == -1){
287
g_err << "Failed to list tables: " << endl
288
<< dict->getNdbError() << endl;
291
for (unsigned i = 0; i < list.count; i++) {
292
NdbDictionary::Dictionary::List::Element& elt = list.elements[i];
294
case NdbDictionary::Object::SystemTable:
295
case NdbDictionary::Object::UserTable:
296
g_ndb->setDatabaseName(elt.database);
297
g_ndb->setSchemaName(elt.schema);
298
if(dict->dropTable(elt.name) != 0){
299
g_err << "Failed to drop table: "
300
<< elt.database << "/" << elt.schema << "/" << elt.name <<endl;
301
g_err << dict->getNdbError() << endl;
305
case NdbDictionary::Object::UniqueHashIndex:
306
case NdbDictionary::Object::OrderedIndex:
307
case NdbDictionary::Object::HashIndexTrigger:
308
case NdbDictionary::Object::IndexTrigger:
309
case NdbDictionary::Object::SubscriptionTrigger:
310
case NdbDictionary::Object::ReadOnlyConstraint:
316
g_ndb->setDatabaseName(db.c_str());
317
g_ndb->setSchemaName(schema.c_str());
322
static int load_table()
324
UtilTransactions clear(* g_table);
325
require(!clear.clearTable(g_ndb));
327
HugoOperations ops(* g_table);
328
require(!ops.startTransaction(g_ndb));
331
size_t uncommitted = 0;
332
bool prepared = false;
333
for(size_t i = 0; i<g_rows; i++){
334
for(op %= OP_COUNT; !((1 << op) & g_use_ops); op = (op + 1) % OP_COUNT);
335
g_ops[i] = g_op_types[op++];
336
if(g_ops[i].start_row){
337
g_ops[i].curr_row = true;
338
g_ops[i].val = rand();
339
require(!ops.pkInsertRecord(g_ndb, i, 1, g_ops[i].val));
342
g_ops[i].curr_row = false;
344
if(uncommitted >= 100){
345
require(!ops.execute_Commit(g_ndb));
346
require(!ops.getTransaction()->restart());
352
require(!ops.execute_Commit(g_ndb));
354
require(!ops.closeTransaction(g_ndb));
356
g_info << "Inserted " << rows << " rows" << endl;
360
static int pause_lcp(int error)
362
int nodes = g_restarter.getNumDbNodes();
364
int filter[] = { 15, NDB_MGM_EVENT_CATEGORY_INFO, 0 };
365
int fd = ndb_mgm_listen_event(g_restarter.handle, filter);
367
require(!g_restarter.insertErrorInAllNodes(error));
368
int dump[] = { DumpStateOrd::DihStartLcpImmediately };
369
require(!g_restarter.dumpStateAllNodes(dump, 1));
373
SocketInputStream in(fd, 1000);
376
tmp = in.gets(buf, 1024);
380
if(sscanf(tmp, "%*[^:]: LCP: %d ", &id) == 1 && id == error &&
386
} while(count++ < 30);
392
static int do_op(int row)
394
HugoOperations & ops = * g_hugo_ops;
395
if(strcmp(g_ops[row].op1, "INS") == 0){
396
require(!g_ops[row].curr_row);
397
g_ops[row].curr_row = true;
398
g_ops[row].val = rand();
399
require(!ops.pkInsertRecord(g_ndb, row, 1, g_ops[row].val));
400
} else if(strcmp(g_ops[row].op1, "UPD") == 0){
401
require(g_ops[row].curr_row);
402
g_ops[row].val = rand();
403
require(!ops.pkUpdateRecord(g_ndb, row, 1, g_ops[row].val));
404
} else if(strcmp(g_ops[row].op1, "DEL") == 0){
405
require(g_ops[row].curr_row);
406
g_ops[row].curr_row = false;
407
require(!ops.pkDeleteRecord(g_ndb, row, 1));
410
require(!ops.execute_NoCommit(g_ndb));
412
if(g_ops[row].op2 == 0){
413
} else if(strcmp(g_ops[row].op2, "INS") == 0){
414
require(!g_ops[row].curr_row);
415
g_ops[row].curr_row = true;
416
g_ops[row].val = rand();
417
require(!ops.pkInsertRecord(g_ndb, row, 1, g_ops[row].val));
418
} else if(strcmp(g_ops[row].op2, "UPD") == 0){
419
require(g_ops[row].curr_row);
420
g_ops[row].val = rand();
421
require(!ops.pkUpdateRecord(g_ndb, row, 1, g_ops[row].val));
422
} else if(strcmp(g_ops[row].op2, "DEL") == 0){
423
require(g_ops[row].curr_row);
424
g_ops[row].curr_row = false;
425
require(!ops.pkDeleteRecord(g_ndb, row, 1));
428
if(g_ops[row].op2 != 0)
429
require(!ops.execute_NoCommit(g_ndb));
431
if(g_ops[row].op3 == 0){
432
} else if(strcmp(g_ops[row].op3, "INS") == 0){
433
require(!g_ops[row].curr_row);
434
g_ops[row].curr_row = true;
435
g_ops[row].val = rand();
436
require(!ops.pkInsertRecord(g_ndb, row, 1, g_ops[row].val));
437
} else if(strcmp(g_ops[row].op3, "UPD") == 0){
438
require(g_ops[row].curr_row);
439
g_ops[row].val = rand();
440
require(!ops.pkUpdateRecord(g_ndb, row, 1, g_ops[row].val));
441
} else if(strcmp(g_ops[row].op3, "DEL") == 0){
442
require(g_ops[row].curr_row);
443
g_ops[row].curr_row = false;
444
require(!ops.pkDeleteRecord(g_ndb, row, 1));
447
if(g_ops[row].op3 != 0)
448
require(!ops.execute_NoCommit(g_ndb));
453
static int continue_lcp(int error)
455
int filter[] = { 15, NDB_MGM_EVENT_CATEGORY_INFO, 0 };
458
fd = ndb_mgm_listen_event(g_restarter.handle, filter);
462
int args[] = { DumpStateOrd::LCPContinue };
463
if(g_restarter.dumpStateAllNodes(args, 1) != 0)
469
SocketInputStream in(fd, 1000);
471
int nodes = g_restarter.getNumDbNodes();
473
tmp = in.gets(buf, 1024);
477
if(sscanf(tmp, "%*[^:]: LCP: %d ", &id) == 1 && id == error &&
483
} while(count++ < 30);
492
HugoOperations & ops = * g_hugo_ops;
493
int res = ops.execute_Commit(g_ndb);
495
return ops.getTransaction()->restart();
502
g_info << "Restarting cluster" << endl;
503
g_hugo_ops->closeTransaction(g_ndb);
507
require(!g_restarter.restartAll());
508
require(!g_restarter.waitClusterStarted(30));
509
require(!connect_ndb());
511
g_table = g_ndb->getDictionary()->getTable(g_tablename);
513
require(g_hugo_ops = new HugoOperations(* g_table));
514
require(!g_hugo_ops->startTransaction(g_ndb));
518
static int validate()
520
HugoOperations ops(* g_table);
521
for(size_t i = 0; i<g_rows; i++){
522
require(g_ops[i].curr_row == g_ops[i].end_row);
523
require(!ops.startTransaction(g_ndb));
524
ops.pkReadRecord(g_ndb, i, 1);
525
int res = ops.execute_Commit(g_ndb);
526
if(g_ops[i].curr_row){
527
require(res == 0 && ops.verifyUpdatesValue(g_ops[i].val) == 0);
531
ops.closeTransaction(g_ndb);
534
for(size_t j = 0; j<10; j++){
535
UtilTransactions clear(* g_table);
536
require(!clear.clearTable(g_ndb));
538
HugoTransactions trans(* g_table);
539
require(trans.loadTable(g_ndb, 1024) == 0);