« back to all changes in this revision
Viewing changes to drizzled/plugin/storage_engine.cc
- browse files at revision 1273.13.92
- compare with another revision
- compare with revision 722.1.8
- download diff
- download tarball
- view history from revision 1273.13.92
- Committer: Brian Aker
- Date: 2010-02-19 16:11:03 UTC
- mfrom: (1273.1.19 refactor-engine-to-resource-manager)
- mto: (1273.19.13 fix_is)
- mto: This revision was merged to the branch mainline in revision 1305.
- Revision ID: brian@gaz-20100219161103-yacb9jzu2eqiubyp
Merge Jay.
- files added:
- drizzled/plugin/transactional_storage_engine.cc
- files renamed:
- drizzled/ha_trx_info.cc => drizzled/resource_context.cc
- drizzled/ha_trx_info.h => drizzled/resource_context.h
- files modified:
- drizzled/cursor.cc
added
removed
drizzled/plugin/storage_engine.cc
50
50
#include "drizzled/internal/my_sys.h"
51
51
#include "drizzled/db.h"
52
52
53
54
53
#include <drizzled/table_proto.h>
55
54
56
#include "drizzled/hash.h"
57
58
55
static bool shutdown_has_begun= false; // Once we put in the container for the vector/etc for engines this will go away.
59
56
60
57
using namespace std;
62
59
namespace drizzled
63
60
{
64
61
65
typedef hash_map<std::string, plugin::StorageEngine *> EngineMap;
66
typedef std::vector<plugin::StorageEngine *> EngineVector;
62
namespace plugin
63
{
67
64
68
65
static EngineVector vector_of_engines;
69
static EngineVector vector_of_transactional_engines;
70
66
71
const std::string plugin::UNKNOWN_STRING("UNKNOWN");
72
const std::string plugin::DEFAULT_DEFINITION_FILE_EXT(".dfe");
67
const std::string UNKNOWN_STRING("UNKNOWN");
68
const std::string DEFAULT_DEFINITION_FILE_EXT(".dfe");
73
69
74
70
static std::set<std::string> set_of_table_definition_ext;
75
71
76
plugin::StorageEngine::StorageEngine(const string name_arg,
77
const bitset<HTON_BIT_SIZE> &flags_arg,
78
bool support_2pc)
72
StorageEngine::StorageEngine(const string name_arg,
73
const bitset<HTON_BIT_SIZE> &flags_arg)
79
74
: Plugin(name_arg, "StorageEngine"),
80
two_phase_commit(support_2pc),
81
75
enabled(true),
82
76
flags(flags_arg),
83
77
slot(0)
85
79
if (enabled)
86
80
{
87
81
slot= total_ha++;
88
if (two_phase_commit)
89
total_ha_2pc++;
90
82
}
91
83
pthread_mutex_init(&proto_cache_mutex, NULL);
92
84
}
93
85
94
95
plugin::StorageEngine::~StorageEngine()
86
StorageEngine::~StorageEngine()
96
87
{
97
88
pthread_mutex_destroy(&proto_cache_mutex);
98
89
}
99
90
100
void plugin::StorageEngine::setTransactionReadWrite(Session& session)
91
void StorageEngine::setTransactionReadWrite(Session& session)
101
92
{
102
Ha_trx_info *ha_info= session.getEngineInfo(this);
103
104
/*
105
When a storage engine method is called, the transaction must
106
have been started, unless it's a DDL call, for which the
107
storage engine starts the transaction internally, and commits
108
it internally, without registering in the ha_list.
109
Unfortunately here we can't know know for sure if the engine
110
has registered the transaction or not, so we must check.
111
*/
112
if (ha_info->is_started())
113
{
114
/*
115
* table_share can be NULL in plugin::StorageEngine::dropTable().
116
*/
117
ha_info->set_trx_read_write();
118
}
93
TransactionContext &statement_ctx= session.transaction.stmt;
94
statement_ctx.markModifiedNonTransData();
119
95
}
120
96
121
122
123
int plugin::StorageEngine::doRenameTable(Session *,
97
int StorageEngine::doRenameTable(Session *,
124
98
const char *from,
125
99
const char *to)
126
100
{
153
127
@retval
154
128
!0 Error
155
129
*/
156
int plugin::StorageEngine::doDropTable(Session&,
130
int StorageEngine::doDropTable(Session&,
157
131
const string table_path)
158
132
{
159
133
int error= 0;
176
150
return error;
177
151
}
178
152
179
const char *plugin::StorageEngine::checkLowercaseNames(const char *path,
153
const char *StorageEngine::checkLowercaseNames(const char *path,
180
154
char *tmp_path)
181
155
{
182
156
if (flags.test(HTON_BIT_FILE_BASED))
199
173
}
200
174
201
175
202
bool plugin::StorageEngine::addPlugin(plugin::StorageEngine *engine)
176
bool StorageEngine::addPlugin(StorageEngine *engine)
203
177
{
204
178
205
179
vector_of_engines.push_back(engine);
206
180
207
if (engine->check_flag(HTON_BIT_DOES_TRANSACTIONS))
208
vector_of_transactional_engines.push_back(engine);
209
210
181
if (engine->getTableDefinitionFileExtension().length())
211
182
{
212
183
assert(engine->getTableDefinitionFileExtension().length() == DEFAULT_DEFINITION_FILE_EXT.length());
216
187
return false;
217
188
}
218
189
219
void plugin::StorageEngine::removePlugin(plugin::StorageEngine *)
190
void StorageEngine::removePlugin(StorageEngine *)
220
191
{
221
192
if (shutdown_has_begun == false)
222
193
{
223
194
vector_of_engines.clear();
224
vector_of_transactional_engines.clear();
225
195
226
196
shutdown_has_begun= true;
227
197
}
228
198
}
229
199
230
200
class FindEngineByName
231
: public unary_function<plugin::StorageEngine *, bool>
201
: public unary_function<StorageEngine *, bool>
232
202
{
233
203
const string target;
234
204
public:
245
215
}
246
216
};
247
217
248
plugin::StorageEngine *plugin::StorageEngine::findByName(string find_str)
218
StorageEngine *StorageEngine::findByName(string find_str)
249
219
{
250
220
transform(find_str.begin(), find_str.end(),
251
221
find_str.begin(), ::tolower);
264
234
return NULL;
265
235
}
266
236
267
plugin::StorageEngine *plugin::StorageEngine::findByName(Session& session,
237
StorageEngine *StorageEngine::findByName(Session& session,
268
238
string find_str)
269
239
{
270
240
288
258
}
289
259
290
260
class StorageEngineCloseConnection
291
: public unary_function<plugin::StorageEngine *, void>
261
: public unary_function<StorageEngine *, void>
292
262
{
293
263
Session *session;
294
264
public:
308
278
@note
309
279
don't bother to rollback here, it's done already
310
280
*/
311
void plugin::StorageEngine::closeConnection(Session* session)
281
void StorageEngine::closeConnection(Session* session)
312
282
{
313
283
for_each(vector_of_engines.begin(), vector_of_engines.end(),
314
284
StorageEngineCloseConnection(session));
315
285
}
316
286
317
void plugin::StorageEngine::dropDatabase(char* path)
287
void StorageEngine::dropDatabase(char* path)
318
288
{
319
289
for_each(vector_of_engines.begin(), vector_of_engines.end(),
320
bind2nd(mem_fun(&plugin::StorageEngine::drop_database),path));
321
}
322
323
int plugin::StorageEngine::commitOrRollbackByXID(XID *xid, bool commit)
324
{
325
vector<int> results;
326
327
if (commit)
328
transform(vector_of_engines.begin(), vector_of_engines.end(), results.begin(),
329
bind2nd(mem_fun(&plugin::StorageEngine::commit_by_xid),xid));
330
else
331
transform(vector_of_engines.begin(), vector_of_engines.end(), results.begin(),
332
bind2nd(mem_fun(&plugin::StorageEngine::rollback_by_xid),xid));
333
334
if (find_if(results.begin(), results.end(), bind2nd(equal_to<int>(),0))
335
== results.end())
336
return 1;
337
return 0;
338
}
339
340
/**
341
@details
342
This function should be called when MySQL sends rows of a SELECT result set
343
or the EOF mark to the client. It releases a possible adaptive hash index
344
S-latch held by session in InnoDB and also releases a possible InnoDB query
345
FIFO ticket to enter InnoDB. To save CPU time, InnoDB allows a session to
346
keep them over several calls of the InnoDB Cursor interface when a join
347
is executed. But when we let the control to pass to the client they have
348
to be released because if the application program uses mysql_use_result(),
349
it may deadlock on the S-latch if the application on another connection
350
performs another SQL query. In MySQL-4.1 this is even more important because
351
there a connection can have several SELECT queries open at the same time.
352
353
@param session the thread handle of the current connection
354
355
@return
356
always 0
357
*/
358
int plugin::StorageEngine::releaseTemporaryLatches(Session *session)
359
{
360
for_each(vector_of_transactional_engines.begin(), vector_of_transactional_engines.end(),
361
bind2nd(mem_fun(&plugin::StorageEngine::release_temporary_latches),session));
362
return 0;
363
}
364
365
bool plugin::StorageEngine::flushLogs(plugin::StorageEngine *engine)
290
bind2nd(mem_fun(&StorageEngine::drop_database),path));
291
}
292
293
bool StorageEngine::flushLogs(StorageEngine *engine)
366
294
{
367
295
if (engine == NULL)
368
296
{
369
297
if (find_if(vector_of_engines.begin(), vector_of_engines.end(),
370
mem_fun(&plugin::StorageEngine::flush_logs))
298
mem_fun(&StorageEngine::flush_logs))
371
299
!= vector_of_engines.begin())
372
300
return true;
373
301
}
380
308
return false;
381
309
}
382
310
383
/**
384
recover() step of xa.
385
386
@note
387
there are three modes of operation:
388
- automatic recover after a crash
389
in this case commit_list != 0, tc_heuristic_recover==0
390
all xids from commit_list are committed, others are rolled back
391
- manual (heuristic) recover
392
in this case commit_list==0, tc_heuristic_recover != 0
393
DBA has explicitly specified that all prepared transactions should
394
be committed (or rolled back).
395
- no recovery (MySQL did not detect a crash)
396
in this case commit_list==0, tc_heuristic_recover == 0
397
there should be no prepared transactions in this case.
398
*/
399
class XARecover : unary_function<plugin::StorageEngine *, void>
400
{
401
int trans_len, found_foreign_xids, found_my_xids;
402
bool result;
403
XID *trans_list;
404
HASH *commit_list;
405
bool dry_run;
406
public:
407
XARecover(XID *trans_list_arg, int trans_len_arg,
408
HASH *commit_list_arg, bool dry_run_arg)
409
: trans_len(trans_len_arg), found_foreign_xids(0), found_my_xids(0),
410
result(false),
411
trans_list(trans_list_arg), commit_list(commit_list_arg),
412
dry_run(dry_run_arg)
413
{}
414
415
int getForeignXIDs()
416
{
417
return found_foreign_xids;
418
}
419
420
int getMyXIDs()
421
{
422
return found_my_xids;
423
}
424
425
result_type operator() (argument_type engine)
426
{
427
428
int got;
429
430
if (engine->is_enabled())
431
{
432
while ((got= engine->recover(trans_list, trans_len)) > 0 )
433
{
434
errmsg_printf(ERRMSG_LVL_INFO,
435
_("Found %d prepared transaction(s) in %s"),
436
got, engine->getName().c_str());
437
for (int i=0; i < got; i ++)
438
{
439
my_xid x=trans_list[i].get_my_xid();
440
if (!x) // not "mine" - that is generated by external TM
441
{
442
xid_cache_insert(trans_list+i, XA_PREPARED);
443
found_foreign_xids++;
444
continue;
445
}
446
if (dry_run)
447
{
448
found_my_xids++;
449
continue;
450
}
451
// recovery mode
452
if (commit_list ?
453
hash_search(commit_list, (unsigned char *)&x, sizeof(x)) != 0 :
454
tc_heuristic_recover == TC_HEURISTIC_RECOVER_COMMIT)
455
{
456
engine->commit_by_xid(trans_list+i);
457
}
458
else
459
{
460
engine->rollback_by_xid(trans_list+i);
461
}
462
}
463
if (got < trans_len)
464
break;
465
}
466
}
467
}
468
};
469
470
int plugin::StorageEngine::recover(HASH *commit_list)
471
{
472
XID *trans_list= NULL;
473
int trans_len= 0;
474
475
bool dry_run= (commit_list==0 && tc_heuristic_recover==0);
476
477
/* commit_list and tc_heuristic_recover cannot be set both */
478
assert(commit_list==0 || tc_heuristic_recover==0);
479
480
/* if either is set, total_ha_2pc must be set too */
481
if (total_ha_2pc <= 1)
482
return 0;
483
484
485
#ifndef WILL_BE_DELETED_LATER
486
487
/*
488
for now, only InnoDB supports 2pc. It means we can always safely
489
rollback all pending transactions, without risking inconsistent data
490
*/
491
492
assert(total_ha_2pc == 2); // only InnoDB and binlog
493
tc_heuristic_recover= TC_HEURISTIC_RECOVER_ROLLBACK; // forcing ROLLBACK
494
dry_run=false;
495
#endif
496
for (trans_len= MAX_XID_LIST_SIZE ;
497
trans_list==0 && trans_len > MIN_XID_LIST_SIZE; trans_len/=2)
498
{
499
trans_list=(XID *)malloc(trans_len*sizeof(XID));
500
}
501
if (!trans_list)
502
{
503
errmsg_printf(ERRMSG_LVL_ERROR, ER(ER_OUTOFMEMORY), trans_len*sizeof(XID));
504
return(1);
505
}
506
507
if (commit_list)
508
errmsg_printf(ERRMSG_LVL_INFO, _("Starting crash recovery..."));
509
510
511
XARecover recover_func(trans_list, trans_len, commit_list, dry_run);
512
for_each(vector_of_transactional_engines.begin(), vector_of_transactional_engines.end(),
513
recover_func);
514
free(trans_list);
515
516
if (recover_func.getForeignXIDs())
517
errmsg_printf(ERRMSG_LVL_WARN,
518
_("Found %d prepared XA transactions"),
519
recover_func.getForeignXIDs());
520
if (dry_run && recover_func.getMyXIDs())
521
{
522
errmsg_printf(ERRMSG_LVL_ERROR,
523
_("Found %d prepared transactions! It means that drizzled "
524
"was not shut down properly last time and critical "
525
"recovery information (last binlog or %s file) was "
526
"manually deleted after a crash. You have to start "
527
"drizzled with the --tc-heuristic-recover switch to "
528
"commit or rollback pending transactions."),
529
recover_func.getMyXIDs(), opt_tc_log_file);
530
return(1);
531
}
532
if (commit_list)
533
errmsg_printf(ERRMSG_LVL_INFO, _("Crash recovery finished."));
534
return(0);
535
}
536
537
int plugin::StorageEngine::startConsistentSnapshot(Session *session)
538
{
539
for_each(vector_of_engines.begin(), vector_of_engines.end(),
540
bind2nd(mem_fun(&plugin::StorageEngine::start_consistent_snapshot),
541
session));
542
return 0;
543
}
544
545
class StorageEngineGetTableDefinition: public unary_function<plugin::StorageEngine *,bool>
311
class StorageEngineGetTableDefinition: public unary_function<StorageEngine *,bool>
546
312
{
547
313
Session& session;
548
314
const char* path;
611
377
to ask engine if there are any new tables that should be written to disk
612
378
or any dropped tables that need to be removed from disk
613
379
*/
614
int plugin::StorageEngine::getTableDefinition(Session& session,
380
int StorageEngine::getTableDefinition(Session& session,
615
381
TableIdentifier &identifier,
616
382
message::Table *table_proto)
617
383
{
620
386
table_proto);
621
387
}
622
388
623
int plugin::StorageEngine::getTableDefinition(Session& session,
389
int StorageEngine::getTableDefinition(Session& session,
624
390
const char* path,
625
391
const char *,
626
392
const char *,
629
395
{
630
396
int err= ENOENT;
631
397
632
vector<plugin::StorageEngine *>::iterator iter=
398
vector<StorageEngine *>::iterator iter=
633
399
find_if(vector_of_engines.begin(), vector_of_engines.end(),
634
400
StorageEngineGetTableDefinition(session, path, NULL, NULL, true, table_proto, &err));
635
401
693
459
/**
694
460
returns ENOENT if the file doesn't exists.
695
461
*/
696
int plugin::StorageEngine::dropTable(Session& session,
462
int StorageEngine::dropTable(Session& session,
697
463
TableIdentifier &identifier,
698
464
bool generate_warning)
699
465
{
700
466
int error= 0;
701
467
int error_proto;
702
468
message::Table src_proto;
703
plugin::StorageEngine* engine;
469
StorageEngine* engine;
704
470
705
error_proto= plugin::StorageEngine::getTableDefinition(session,
471
error_proto= StorageEngine::getTableDefinition(session,
706
472
identifier,
707
473
&src_proto);
708
474
713
479
return ER_CORRUPT_TABLE_DEFINITION;
714
480
}
715
481
716
engine= plugin::StorageEngine::findByName(session,
717
src_proto.engine().name());
482
engine= StorageEngine::findByName(session, src_proto.engine().name());
718
483
719
484
if (engine)
720
485
{
784
549
785
550
@todo refactor to remove goto
786
551
*/
787
int plugin::StorageEngine::createTable(Session& session,
552
int StorageEngine::createTable(Session& session,
788
553
TableIdentifier &identifier,
789
554
bool update_create_info,
790
555
message::Table& table_proto, bool proto_used)
864
629
return(error != 0);
865
630
}
866
631
867
Cursor *plugin::StorageEngine::getCursor(TableShare &share, memory::Root *alloc)
632
Cursor *StorageEngine::getCursor(TableShare &share, memory::Root *alloc)
868
633
{
869
634
assert(enabled);
870
635
return create(share, alloc);
873
638
/**
874
639
TODO -> Remove this to force all engines to implement their own file. Solves the "we only looked at dfe" problem.
875
640
*/
876
void plugin::StorageEngine::doGetTableNames(CachedDirectory &directory, string&, set<string>& set_of_names)
641
void StorageEngine::doGetTableNames(CachedDirectory &directory, string&, set<string>& set_of_names)
877
642
{
878
643
CachedDirectory::Entries entries= directory.getEntries();
879
644
904
669
}
905
670
906
671
class AddTableName :
907
public unary_function<plugin::StorageEngine *, void>
672
public unary_function<StorageEngine *, void>
908
673
{
909
674
string db;
910
675
CachedDirectory& directory;
925
690
}
926
691
};
927
692
928
void plugin::StorageEngine::getSchemaNames(set<string>& set_of_names)
693
void StorageEngine::getSchemaNames(set<string>& set_of_names)
929
694
{
930
695
CachedDirectory directory(drizzle_data_home, CachedDirectory::DIRECTORY);
931
696
950
715
/*
951
716
Return value is "if parsed"
952
717
*/
953
bool plugin::StorageEngine::getSchemaDefinition(const std::string &schema_name, message::Schema &proto)
718
bool StorageEngine::getSchemaDefinition(const std::string &schema_name, message::Schema &proto)
954
719
{
955
720
int ret;
956
721
968
733
return ret == 0 ? true : false;
969
734
}
970
735
971
void plugin::StorageEngine::getTableNames(const string& db, set<string>& set_of_names)
736
void StorageEngine::getTableNames(const string& db, set<string>& set_of_names)
972
737
{
973
738
char tmp_path[FN_REFLEN];
974
739
994
759
}
995
760
996
761
/* This will later be converted to TableIdentifiers */
997
class DropTables: public unary_function<plugin::StorageEngine *, void>
762
class DropTables: public unary_function<StorageEngine *, void>
998
763
{
999
764
Session &session;
1000
765
set<string>& set_of_names;
1028
793
1029
794
Note-> Unlike MySQL, we do not, on purpose, delete files that do not match any engines.
1030
795
*/
1031
void plugin::StorageEngine::removeLostTemporaryTables(Session &session, const char *directory)
796
void StorageEngine::removeLostTemporaryTables(Session &session, const char *directory)
1032
797
{
1033
798
CachedDirectory dir(directory, set_of_table_definition_ext);
1034
799
set<string> set_of_table_names;
1071
836
*/
1072
837
set<string> all_exts= set_of_table_definition_ext;
1073
838
1074
for (vector<plugin::StorageEngine *>::iterator iter= vector_of_engines.begin();
839
for (vector<StorageEngine *>::iterator iter= vector_of_engines.begin();
1075
840
iter != vector_of_engines.end() ; iter++)
1076
841
{
1077
842
for (const char **ext= (*iter)->bas_ext(); *ext ; ext++)
1105
870
- table->s->path
1106
871
- table->alias
1107
872
*/
1108
void plugin::StorageEngine::print_error(int error, myf errflag, Table &table)
873
void StorageEngine::print_error(int error, myf errflag, Table &table)
1109
874
{
1110
875
print_error(error, errflag, &table);
1111
876
}
1112
877
1113
void plugin::StorageEngine::print_error(int error, myf errflag, Table *table)
878
void StorageEngine::print_error(int error, myf errflag, Table *table)
1114
879
{
1115
880
int textno= ER_GET_ERRNO;
1116
881
switch (error) {
1322
1087
@return
1323
1088
Returns true if this is a temporary error
1324
1089
*/
1325
bool plugin::StorageEngine::get_error_message(int , String* )
1090
bool StorageEngine::get_error_message(int , String* )
1326
1091
{
1327
1092
return false;
1328
1093
}
1329
1094
1330
1095
1331
void plugin::StorageEngine::print_keydup_error(uint32_t key_nr, const char *msg, Table &table)
1096
void StorageEngine::print_keydup_error(uint32_t key_nr, const char *msg, Table &table)
1332
1097
{
1333
1098
/* Write the duplicated key in the error message */
1334
1099
char key[MAX_KEY_LENGTH];
1356
1121
}
1357
1122
1358
1123
1359
int plugin::StorageEngine::deleteDefinitionFromPath(TableIdentifier &identifier)
1124
int StorageEngine::deleteDefinitionFromPath(TableIdentifier &identifier)
1360
1125
{
1361
1126
string path(identifier.getPath());
1362
1127
1365
1130
return internal::my_delete(path.c_str(), MYF(0));
1366
1131
}
1367
1132
1368
int plugin::StorageEngine::renameDefinitionFromPath(TableIdentifier &dest, TableIdentifier &src)
1133
int StorageEngine::renameDefinitionFromPath(TableIdentifier &dest, TableIdentifier &src)
1369
1134
{
1370
1135
string src_path(src.getPath());
1371
1136
string dest_path(dest.getPath());
1376
1141
return internal::my_rename(src_path.c_str(), dest_path.c_str(), MYF(MY_WME));
1377
1142
}
1378
1143
1379
int plugin::StorageEngine::writeDefinitionFromPath(TableIdentifier &identifier, message::Table &table_proto)
1144
int StorageEngine::writeDefinitionFromPath(TableIdentifier &identifier, message::Table &table_proto)
1380
1145
{
1381
1146
string file_name(identifier.getPath());
1382
1147
1402
1167
return 0;
1403
1168
}
1404
1169
1405
1170
} /* namespace plugin */
1406
1171
} /* namespace drizzled */
Loggerhead is a web-based interface for Breezy
Version: 2.0.1