« back to all changes in this revision
Viewing changes to plugin/pbms/src/database_ms.cc
- browse files at revision 1644
- compare with another revision
- compare with revision 77.1.47
- download diff
- download tarball
- view history from revision 1644
- Committer: Monty Taylor
- Date: 2010-07-06 00:44:32 UTC
- mfrom: (1643.1.13 build)
- Revision ID: mordred@inaugust.com-20100706004432-843uftc92rc2497l
Merged in PBMS, translation updates, a few build fixes and a few bug fixes.
- files added:
- m4/pandora_have_libboost_thread.m4
- plugin/pbms
- plugin/pbms/src
- plugin/pbms/src/cslib
- files modified:
- .quickly
added
removed
plugin/pbms/src/database_ms.cc
1
/* Copyright (c) 2008 PrimeBase Technologies GmbH, Germany
2
*
3
* PrimeBase Media Stream for MySQL
4
*
5
* This program is free software; you can redistribute it and/or modify
6
* it under the terms of the GNU General Public License as published by
7
* the Free Software Foundation; either version 2 of the License, or
8
* (at your option) any later version.
9
*
10
* This program is distributed in the hope that it will be useful,
11
* but WITHOUT ANY WARRANTY; without even the implied warranty of
12
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13
* GNU General Public License for more details.
14
*
15
* You should have received a copy of the GNU General Public License
16
* along with this program; if not, write to the Free Software
17
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18
*
19
* Original author: Paul McCullagh
20
* Continued development: Barry Leslie
21
*
22
* 2007-05-25
23
*
24
* H&G2JCtL
25
*
26
* Network interface.
27
*
28
*/
29
30
#ifdef DRIZZLED
31
#include "config.h"
32
#include <drizzled/common.h>
33
#include <drizzled/session.h>
34
#include <drizzled/table.h>
35
#include <drizzled/message/table.pb.h>
36
#include "drizzled/charset_info.h"
37
#include <drizzled/table_proto.h>
38
#include <drizzled/field.h>
39
#endif
40
41
#include "cslib/CSConfig.h"
42
43
#include <string.h>
44
#include <ctype.h>
45
#include <inttypes.h>
46
47
#include "string.h"
48
49
#include "cslib/CSGlobal.h"
50
#include "cslib/CSLog.h"
51
#include "cslib/CSDirectory.h"
52
#include "cslib/CSStrUtil.h"
53
54
#include "database_ms.h"
55
#include "open_table_ms.h"
56
#include "backup_ms.h"
57
#include "table_ms.h"
58
#include "temp_log_ms.h"
59
#include "network_ms.h"
60
#include "mysql_ms.h"
61
#include "pbmslib.h"
62
#include "transaction_ms.h"
63
//#include "systab_variable_ms.h"
64
#include "systab_httpheader_ms.h"
65
#include "parameters_ms.h"
66
#include "pbmsdaemon_ms.h"
67
68
69
70
CSSyncSortedList *MSDatabase::gDatabaseList;
71
CSSparseArray *MSDatabase::gDatabaseArray;
72
/*
73
* -------------------------------------------------------------------------
74
* PBMS DATABASES
75
*/
76
77
MSDatabase::MSDatabase():
78
myIsPBMS(false),
79
myDatabaseID(0),
80
myDatabaseName(NULL),
81
myDatabasePath(NULL),
82
myTempLogArray(NULL),
83
myCompactorThread(NULL),
84
myTempLogThread(NULL),
85
myRepostoryList(NULL),
86
myBlobCloud(NULL),
87
myBlobType(MS_STANDARD_STORAGE),
88
isBackup(false),
89
iBackupThread(NULL),
90
iBackupTime(0),
91
iRecovering(false),
92
#ifdef HAVE_ALIAS_SUPPORT
93
iBlobAliases(NULL),
94
#endif
95
iClosing(false),
96
iTableList(NULL),
97
iTableArray(NULL),
98
iMaxTableID(0),
99
iWriteTempLog(NULL),
100
iDropping(false),
101
iNextBlobRefId(0)
102
{
103
//startTracking();
104
}
105
106
MSDatabase::~MSDatabase()
107
{
108
iClosing = true;
109
if (iBackupThread) {
110
iBackupThread->stop();
111
iBackupThread->release();
112
iBackupThread = NULL;
113
}
114
115
if (myTempLogThread) {
116
myTempLogThread->stop();
117
myTempLogThread->release();
118
myTempLogThread = NULL;
119
}
120
if (myCompactorThread) {
121
myRepostoryList->wakeup(); // The compator thread waits on this.
122
myCompactorThread->stop();
123
myCompactorThread->release();
124
myCompactorThread = NULL;
125
}
126
127
if (myDatabasePath)
128
myDatabasePath->release();
129
130
if (myDatabaseName)
131
myDatabaseName->release();
132
133
iWriteTempLog = NULL;
134
if (myTempLogArray) {
135
myTempLogArray->clear();
136
myTempLogArray->release();
137
}
138
if (iTableList) {
139
iTableList->clear();
140
iTableList->release();
141
}
142
if (iTableArray){
143
iTableArray->clear();
144
iTableArray->release();
145
}
146
if (myRepostoryList) {
147
myRepostoryList->clear();
148
myRepostoryList->release();
149
}
150
#ifdef HAVE_ALIAS_SUPPORT
151
if (iBlobAliases) {
152
iBlobAliases->ma_close();
153
iBlobAliases->release();
154
}
155
#endif
156
if (myBlobCloud) {
157
myBlobCloud->release();
158
}
159
}
160
161
uint32_t MSDatabase::fileToTableId(const char *file_name, const char *name_part)
162
{
163
uint32_t value = 0;
164
165
if (file_name) {
166
const char *num = file_name + strlen(file_name) - 1;
167
168
while (num >= file_name && *num != '-')
169
num--;
170
if (name_part) {
171
/* Check the name part of the file: */
172
int len = strlen(name_part);
173
174
if (len != num - file_name)
175
return 0;
176
if (strncmp(file_name, name_part, len) != 0)
177
return 0;
178
}
179
num++;
180
if (isdigit(*num))
181
sscanf(num, "%"PRIu32"", &value);
182
}
183
return value;
184
}
185
186
const char *MSDatabase::fileToTableName(size_t size, char *tab_name, const char *file_name)
187
{
188
const char *cptr;
189
size_t len;
190
191
file_name = cs_last_name_of_path(file_name);
192
cptr = file_name + strlen(file_name) - 1;
193
while (cptr > file_name && *cptr != '.')
194
cptr--;
195
if (cptr > file_name && *cptr == '.') {
196
if (strncmp(cptr, ".bs", 2) == 0) {
197
cptr--;
198
while (cptr > file_name && isdigit(*cptr))
199
cptr--;
200
}
201
}
202
203
len = cptr - file_name;
204
if (len > size-1)
205
len = size-1;
206
207
memcpy(tab_name, file_name, len);
208
tab_name[len] = 0;
209
210
/* Return a pointer to what was removed! */
211
return file_name + len;
212
}
213
214
215
const char *MSDatabase::getDatabaseNameCString()
216
{
217
return myDatabaseName->getCString();
218
}
219
220
MSTable *MSDatabase::getTable(CSString *tab_name, bool create)
221
{
222
MSTable *tab;
223
224
enter_();
225
push_(tab_name);
226
lock_(iTableList);
227
if (!(tab = (MSTable *) iTableList->find(tab_name))) {
228
229
if (create) {
230
/* Create a new table: */
231
tab = MSTable::newTable(iMaxTableID+1, RETAIN(tab_name), this, (off64_t) 0, false);
232
iTableList->add(tab);
233
iTableArray->set(iMaxTableID+1, RETAIN(tab));
234
iMaxTableID++;
235
}
236
}
237
if (tab)
238
tab->retain();
239
unlock_(iTableList);
240
release_(tab_name);
241
return_(tab);
242
}
243
244
MSTable *MSDatabase::getTable(const char *tab_name, bool create)
245
{
246
return getTable(CSString::newString(tab_name), create);
247
}
248
249
250
MSTable *MSDatabase::getTable(uint32_t tab_id, bool missing_ok)
251
{
252
MSTable *tab;
253
254
enter_();
255
lock_(iTableList);
256
if (!(tab = (MSTable *) iTableArray->get((uint32_t) tab_id))) {
257
if (missing_ok) {
258
unlock_(iTableList);
259
return_(NULL);
260
}
261
char buffer[CS_EXC_MESSAGE_SIZE];
262
263
cs_strcpy(CS_EXC_MESSAGE_SIZE, buffer, "Unknown table #");
264
cs_strcat(CS_EXC_MESSAGE_SIZE, buffer, (uint32_t) tab_id);
265
cs_strcat(CS_EXC_MESSAGE_SIZE, buffer, " in database ");
266
cs_strcat(CS_EXC_MESSAGE_SIZE, buffer, getDatabaseNameCString());
267
CSException::throwException(CS_CONTEXT, MS_ERR_UNKNOWN_TABLE, buffer);
268
}
269
tab->retain();
270
unlock_(iTableList);
271
return_(tab);
272
}
273
274
MSTable *MSDatabase::getNextTable(uint32_t *pos)
275
{
276
uint32_t i = *pos;
277
MSTable *tab = NULL;
278
279
enter_();
280
lock_(iTableList);
281
while (i < iTableList->getSize()) {
282
tab = (MSTable *) iTableList->itemAt(i++);
283
if (!tab->isToDelete())
284
break;
285
tab = NULL;
286
}
287
if (tab)
288
tab->retain();
289
unlock_(iTableList);
290
*pos = i;
291
return_(tab);
292
}
293
294
void MSDatabase::addTable(uint32_t tab_id, const char *tab_name, off64_t file_size, bool to_delete)
295
{
296
MSTable *tab;
297
298
if (tab_id > iMaxTableID)
299
iMaxTableID = tab_id;
300
tab = MSTable::newTable(tab_id, tab_name, this, file_size, to_delete);
301
iTableList->add(tab);
302
iTableArray->set(tab_id, RETAIN(tab));
303
}
304
305
void MSDatabase::addTableFromFile(CSDirectory *dir, const char *file_name, bool to_delete)
306
{
307
off64_t file_size;
308
uint32_t file_id;
309
char tab_name[MS_TABLE_NAME_SIZE];
310
311
dir->info(NULL, &file_size, NULL);
312
file_id = fileToTableId(file_name);
313
fileToTableName(MS_TABLE_NAME_SIZE, tab_name, file_name);
314
addTable(file_id, tab_name, file_size, to_delete);
315
}
316
317
void MSDatabase::removeTable(MSTable *tab)
318
{
319
enter_();
320
push_(tab);
321
lock_(iTableList);
322
iTableList->remove(tab->myTableName);
323
iTableArray->remove(tab->myTableID);
324
unlock_(iTableList);
325
release_(tab);
326
exit_();
327
}
328
329
void MSDatabase::dropTable(MSTable *tab)
330
{
331
enter_();
332
push_(tab);
333
lock_(iTableList);
334
iTableList->remove(tab->myTableName);
335
iTableArray->remove(tab->myTableID);
336
337
// Cute: you drop the table by adding it with the 'to_delete' flag set to 'true'
338
addTable(tab->myTableID, tab->myTableName->getCString(), tab->getTableFileSize(), true);
339
340
unlock_(iTableList);
341
release_(tab);
342
exit_();
343
}
344
345
// This function is used when dropping tables from a database before
346
// dropping the database itself.
347
CSString *MSDatabase::getATableName()
348
{
349
uint32_t i = 0;
350
MSTable *tab;
351
CSString *name = NULL;
352
353
enter_();
354
lock_(iTableList);
355
356
while ((tab = (MSTable *) iTableList->itemAt(i++)) && tab->isToDelete()) ;
357
if (tab) {
358
name = tab->getTableName();
359
name->retain();
360
}
361
unlock_(iTableList);
362
return_(name);
363
}
364
365
uint32_t MSDatabase::getTableCount()
366
{
367
uint32_t cnt = 0, i = 0;
368
MSTable *tab;
369
370
enter_();
371
lock_(iTableList);
372
373
while ((tab = (MSTable *) iTableList->itemAt(i++))) {
374
if (!tab->isToDelete())
375
cnt++;
376
}
377
378
unlock_(iTableList);
379
return_(cnt);
380
}
381
382
383
void MSDatabase::renameTable(MSTable *tab, const char *to_name)
384
{
385
enter_();
386
lock_(iTableList);
387
iTableList->remove(tab->myTableName);
388
iTableArray->remove(tab->myTableID);
389
390
addTable(tab->myTableID, to_name, tab->getTableFileSize(), false);
391
392
unlock_(iTableList);
393
exit_();
394
}
395
396
void MSDatabase::openWriteRepo(MSOpenTable *otab)
397
{
398
if (otab->myWriteRepo && otab->myWriteRepoFile)
399
return;
400
401
enter_();
402
if (!otab->myWriteRepo)
403
otab->myWriteRepo = lockRepo(0);
404
405
/* Now open the repo file for the open table: */
406
otab->myWriteRepo->openRepoFileForWriting(otab);
407
exit_();
408
}
409
410
MSRepository *MSDatabase::getRepoFullOfTrash(time_t *ret_wait_time)
411
{
412
MSRepository *repo = NULL;
413
time_t wait_time = 0;
414
415
if (ret_wait_time)
416
wait_time = *ret_wait_time;
417
enter_();
418
lock_(myRepostoryList);
419
for (uint32_t i=0; i<myRepostoryList->size(); i++) {
420
retry:
421
if ((repo = (MSRepository *) myRepostoryList->get(i))) {
422
if (!repo->isRemovingFP && !repo->mustBeDeleted && !repo->isRepoLocked()) {
423
if (!repo->myRepoHeadSize) {
424
/* The file has not yet been opened, so the
425
* garbage count will not be known!
426
*/
427
MSRepoFile *repo_file;
428
429
repo->retain();
430
unlock_(myRepostoryList);
431
push_(repo);
432
repo_file = repo->openRepoFile();
433
repo_file->release();
434
release_(repo);
435
lock_(myRepostoryList);
436
goto retry;
437
}
438
if (repo->getGarbageLevel() >= PBMSParameters::getGarbageThreshold()) {
439
/* Make sure there are not temp BLOBs in this repository that have
440
* not yet timed out:
441
*/
442
time_t now = time(NULL);
443
time_t then = repo->myLastTempTime;
444
445
/* Check if there are any temp BLOBs to be removed: */
446
if (now > (time_t)(then + PBMSParameters::getTempBlobTimeout())) {
447
repo->lockRepo(REPO_COMPACTING);
448
repo->retain();
449
break;
450
}
451
else {
452
/* There are temp BLOBs to wait for... */
453
if (!wait_time || wait_time > MSTempLog::adjustWaitTime(then, now))
454
wait_time = MSTempLog::adjustWaitTime(then, now);
455
}
456
}
457
}
458
repo = NULL;
459
}
460
}
461
unlock_(myRepostoryList);
462
if (ret_wait_time)
463
*ret_wait_time = wait_time;
464
return_(repo);
465
}
466
467
MSRepository *MSDatabase::lockRepo(off64_t size)
468
{
469
MSRepository *repo;
470
uint32_t free_slot;
471
472
enter_();
473
lock_(myRepostoryList);
474
free_slot = myRepostoryList->size();
475
/* Find an unlocked repository file that is below the write threshold: */
476
for (uint32_t i=0; i<myRepostoryList->size(); i++) {
477
if ((repo = (MSRepository *) myRepostoryList->get(i))) {
478
if ((!repo->isRepoLocked()) && (!repo->isRemovingFP) && (!repo->mustBeDeleted) &&
479
((repo->myRepoFileSize + size) < PBMSParameters::getRepoThreshold())
480
/**/ && (repo->getGarbageLevel() < PBMSParameters::getGarbageThreshold()))
481
goto found1;
482
}
483
else {
484
if (i < free_slot)
485
free_slot = i;
486
}
487
}
488
489
/* None found, create a new repo file: */
490
new_(repo, MSRepository(free_slot + 1, this, 0));
491
myRepostoryList->set(free_slot, repo);
492
493
found1:
494
repo->retain();
495
repo->lockRepo(REPO_WRITE); // <- The MSRepository::backToPool() will unlock this.
496
unlock_(myRepostoryList);
497
return_(repo);
498
}
499
500
MSRepoFile *MSDatabase::getRepoFileFromPool(uint32_t repo_id, bool missing_ok)
501
{
502
MSRepository *repo;
503
MSRepoFile *file;
504
505
enter_();
506
lock_(myRepostoryList);
507
if (!(repo = (MSRepository *) myRepostoryList->get(repo_id - 1))) {
508
if (!missing_ok) {
509
char buffer[CS_EXC_MESSAGE_SIZE];
510
511
cs_strcpy(CS_EXC_MESSAGE_SIZE, buffer, "Unknown repository file: ");
512
cs_strcat(CS_EXC_MESSAGE_SIZE, buffer, (uint32_t) repo_id);
513
CSException::throwException(CS_CONTEXT, MS_ERR_NOT_FOUND, buffer);
514
}
515
unlock_(myRepostoryList);
516
return_(NULL);
517
}
518
if (repo->isRemovingFP) {
519
char buffer[CS_EXC_MESSAGE_SIZE];
520
521
cs_strcpy(CS_EXC_MESSAGE_SIZE, buffer, "Repository will be removed: ");
522
cs_strcat(CS_EXC_MESSAGE_SIZE, buffer, (uint32_t) repo_id);
523
CSException::throwException(CS_CONTEXT, MS_ERR_REMOVING_REPO, buffer);
524
}
525
repo->retain(); /* Release is here: [++] */
526
file = repo->getRepoFile();
527
unlock_(myRepostoryList);
528
529
if (!file) {
530
file = repo->openRepoFile();
531
lock_(myRepostoryList);
532
repo->addRepoFile(file);
533
file->retain();
534
unlock_(myRepostoryList);
535
}
536
return_(file);
537
}
538
539
void MSDatabase::returnRepoFileToPool(MSRepoFile *file)
540
{
541
MSRepository *repo;
542
543
enter_();
544
lock_(myRepostoryList);
545
push_(file);
546
if ((repo = file->myRepo)) {
547
if (repo->isRemovingFP) {
548
repo->removeRepoFile(file);
549
myRepostoryList->wakeup();
550
}
551
else
552
repo->returnRepoFile(file);
553
repo->release(); /* [++] here is the release. */
554
}
555
release_(file);
556
unlock_(myRepostoryList);
557
exit_();
558
}
559
560
void MSDatabase::removeRepo(uint32_t repo_id, bool *mustQuit)
561
{
562
MSRepository *repo;
563
564
enter_();
565
lock_(myRepostoryList);
566
while ((!mustQuit || !*mustQuit) && !iClosing) {
567
if (!(repo = (MSRepository *) myRepostoryList->get(repo_id - 1)))
568
break;
569
repo->isRemovingFP = true;
570
if (repo->removeRepoFilesNotInUse()) {
571
myRepostoryList->set(repo_id - 1, NULL);
572
break;
573
}
574
/*
575
* Wait for the files that are in use to be
576
* freed.
577
*/
578
myRepostoryList->wait();
579
}
580
unlock_(myRepostoryList);
581
exit_();
582
}
583
584
void MSDatabase::queueTempLogEvent(MSOpenTable *otab, int type, uint32_t tab_id, uint64_t blob_id, uint32_t auth_code,
585
uint32_t *log_id, uint32_t *log_offset, uint32_t *q_time)
586
{
587
MSTempLogItemRec item;
588
uint32_t timev;
589
590
// Each otab object holds an handle to an instance of an OPEN
591
// temp log. This is so that each thread has it's own open temp log
592
// and doesn't need to be opened and close it constantly.
593
594
enter_();
595
lock_(myTempLogArray);
596
if (!iWriteTempLog) {
597
iWriteTempLog = (MSTempLog *) myTempLogArray->last();
598
if (!iWriteTempLog) {
599
new_(iWriteTempLog, MSTempLog(1, this, 0));
600
myTempLogArray->set(1, iWriteTempLog);
601
}
602
}
603
if (!otab->myTempLogFile)
604
otab->myTempLogFile = iWriteTempLog->openTempLog();
605
else if (otab->myTempLogFile->myTempLogID != iWriteTempLog->myLogID) {
606
otab->myTempLogFile->release();
607
otab->myTempLogFile = NULL;
608
otab->myTempLogFile = iWriteTempLog->openTempLog();
609
}
610
611
if (iWriteTempLog->myTempLogSize >= PBMSParameters::getTempLogThreshold()) {
612
uint32_t tmp_log_id = iWriteTempLog->myLogID + 1;
613
614
new_(iWriteTempLog, MSTempLog(tmp_log_id, this, 0));
615
myTempLogArray->set(tmp_log_id, iWriteTempLog);
616
617
otab->myTempLogFile->release();
618
otab->myTempLogFile = NULL;
619
otab->myTempLogFile = iWriteTempLog->openTempLog();
620
621
}
622
623
timev = time(NULL);
624
*log_id = iWriteTempLog->myLogID;
625
*log_offset = (uint32_t) iWriteTempLog->myTempLogSize;
626
if (q_time)
627
*q_time = timev;
628
iWriteTempLog->myTempLogSize += iWriteTempLog->myTemplogRecSize;
629
unlock_(myTempLogArray);
630
631
CS_SET_DISK_1(item.ti_type_1, type);
632
CS_SET_DISK_4(item.ti_table_id_4, tab_id);
633
CS_SET_DISK_6(item.ti_blob_id_6, blob_id);
634
CS_SET_DISK_4(item.ti_auth_code_4, auth_code);
635
CS_SET_DISK_4(item.ti_time_4, timev);
636
otab->myTempLogFile->write(&item, *log_offset, sizeof(MSTempLogItemRec));
637
638
639
exit_();
640
}
641
642
#ifdef HAVE_ALIAS_SUPPORT
643
void MSDatabase::queueForDeletion(MSOpenTable *otab, int type, uint32_t tab_id, uint64_t blob_id, uint32_t auth_code,
644
uint32_t *log_id, uint32_t *log_offset, uint32_t *q_time, MSDiskAliasPtr aliasDiskRec)
645
{
646
enter_();
647
648
queueTempLogEvent(otab, type, tab_id, blob_id, auth_code, log_id, log_offset, q_time);
649
650
// If it has an alias remove it from the ailias index.
651
if (aliasDiskRec) {
652
try_(a) {
653
deleteBlobAlias(aliasDiskRec);
654
}
655
catch_(a);
656
self->logException();
657
cont_(a);
658
}
659
660
exit_();
661
}
662
#endif
663
664
MSTempLogFile *MSDatabase::openTempLogFile(uint32_t log_id, size_t *log_rec_size, size_t *log_head_size)
665
{
666
MSTempLog *log;
667
MSTempLogFile *log_file = NULL;
668
669
enter_();
670
lock_(myTempLogArray);
671
if (log_id)
672
log = (MSTempLog *) myTempLogArray->get(log_id);
673
else
674
log = (MSTempLog *) myTempLogArray->first();
675
if (log) {
676
log_file = log->openTempLog();
677
if (log_rec_size)
678
*log_rec_size = log->myTemplogRecSize;
679
if (log_head_size)
680
*log_head_size = log->myTempLogHeadSize;
681
}
682
unlock_(myTempLogArray);
683
return_(log_file);
684
}
685
686
uint32_t MSDatabase::getTempLogCount()
687
{
688
uint32_t count;
689
690
enter_();
691
lock_(myTempLogArray);
692
count = myTempLogArray->size();
693
unlock_(myTempLogArray);
694
return_(count);
695
}
696
697
void MSDatabase::removeTempLog(uint32_t log_id)
698
{
699
enter_();
700
lock_(myTempLogArray);
701
myTempLogArray->remove(log_id);
702
unlock_(myTempLogArray);
703
exit_();
704
}
705
706
CSObject *MSDatabase::getKey()
707
{
708
return (CSObject *) myDatabaseName;
709
}
710
711
int MSDatabase::compareKey(CSObject *key)
712
{
713
return myDatabaseName->compare((CSString *) key);
714
}
715
716
MSCompactorThread *MSDatabase::getCompactorThread()
717
{
718
return myCompactorThread;
719
}
720
721
CSSyncVector *MSDatabase::getRepositoryList()
722
{
723
return myRepostoryList;
724
}
725
726
#ifdef HAVE_ALIAS_SUPPORT
727
uint32_t MSDatabase::registerBlobAlias(uint32_t repo_id, uint64_t repo_offset, const char *alias)
728
{
729
uint32_t hash;
730
bool can_retry = true;
731
enter_();
732
733
retry:
734
lock_(&iBlobAliaseLock);
735
736
try_(a) {
737
hash = iBlobAliases->addAlias(repo_id, repo_offset, alias);
738
}
739
740
catch_(a) {
741
unlock_(&iBlobAliaseLock);
742
if (can_retry) {
743
// It can be that a duplicater alias exists that was deleted
744
// but the transaction has not been written to the repository yet.
745
// Flush all committed transactions to the repository file.
746
MSTransactionManager::flush();
747
can_retry = false;
748
goto retry;
749
}
750
throw_();
751
}
752
753
cont_(a);
754
unlock_(&iBlobAliaseLock);
755
return_(hash);
756
}
757
758
uint32_t MSDatabase::updateBlobAlias(uint32_t repo_id, uint64_t repo_offset, uint32_t old_alias_hash, const char *alias)
759
{
760
uint32_t new_hash;
761
enter_();
762
lock_(&iBlobAliaseLock);
763
764
new_hash = iBlobAliases->addAlias(repo_id, repo_offset, alias);
765
iBlobAliases->deleteAlias(repo_id, repo_offset, old_alias_hash);
766
767
unlock_(&iBlobAliaseLock);
768
return_(new_hash);
769
}
770
771
void MSDatabase::deleteBlobAlias(MSDiskAliasPtr diskRec)
772
{
773
enter_();
774
lock_(&iBlobAliaseLock);
775
iBlobAliases->deleteAlias(diskRec);
776
unlock_(&iBlobAliaseLock);
777
exit_();
778
}
779
780
void MSDatabase::deleteBlobAlias(uint32_t repo_id, uint64_t repo_offset, uint32_t alias_hash)
781
{
782
MSDiskAliasRec diskRec;
783
784
CS_SET_DISK_4(diskRec.ar_repo_id_4, repo_id);
785
CS_SET_DISK_8(diskRec.ar_offset_8, repo_offset);
786
CS_SET_DISK_4(diskRec.ar_hash_4, alias_hash);
787
deleteBlobAlias(&diskRec);
788
}
789
790
void MSDatabase::moveBlobAlias(uint32_t old_repo_id, uint64_t old_repo_offset, uint32_t alias_hash, uint32_t new_repo_id, uint64_t new_repo_offset)
791
{
792
enter_();
793
lock_(&iBlobAliaseLock);
794
iBlobAliases->resetAlias(old_repo_id, old_repo_offset, alias_hash, new_repo_id, new_repo_offset);
795
unlock_(&iBlobAliaseLock);
796
exit_();
797
}
798
#endif
799
800
bool MSDatabase::isValidHeaderField(const char *name)
801
{
802
bool is_valid = false;
803
CSString *header;
804
enter_();
805
806
if (name && *name) {
807
if (strcasecmp(name, MS_ALIAS_TAG)) {
808
lock_(&iHTTPMetaDataHeaders);
809
header = CSString::newString(name);
810
push_(header);
811
812
is_valid = (iHTTPMetaDataHeaders.find(header) != NULL);
813
release_(header);
814
815
unlock_(&iHTTPMetaDataHeaders);
816
} else
817
is_valid = true;
818
}
819
820
return_(is_valid);
821
}
822
823
void MSDatabase::startUp(const char *default_http_headers)
824
{
825
enter_();
826
827
new_(gDatabaseList, CSSyncSortedList);
828
new_(gDatabaseArray, CSSparseArray(5));
829
MSHTTPHeaderTable::setDefaultMetaDataHeaders(default_http_headers);
830
PBMSSystemTables::systemTablesStartUp();
831
PBMSParameters::setBackupDatabaseID(1);
832
exit_();
833
}
834
835
void MSDatabase::stopThreads()
836
{
837
MSDatabase *db;
838
839
enter_();
840
if (gDatabaseList) {
841
lock_(gDatabaseList);
842
for (int i=0;;i++) {
843
if (!(db = (MSDatabase *) gDatabaseList->itemAt(i)))
844
break;
845
db->iClosing = true;
846
847
if (db->myTempLogThread) {
848
db->myTempLogThread->stop();
849
db->myTempLogThread->release();
850
db->myTempLogThread = NULL;
851
}
852
if (db->myCompactorThread) {
853
db->myRepostoryList->wakeup(); // The compator thread waits on this.
854
db->myCompactorThread->stop();
855
db->myCompactorThread->release();
856
db->myCompactorThread = NULL;
857
}
858
859
if (db->iBackupThread) {
860
db->iBackupThread->stop();
861
db->iBackupThread->release();
862
db->iBackupThread = NULL;
863
}
864
865
}
866
867
unlock_(gDatabaseList);
868
}
869
exit_();
870
}
871
872
void MSDatabase::shutDown()
873
{
874
875
if (gDatabaseArray) {
876
gDatabaseArray->clear();
877
gDatabaseArray->release();
878
gDatabaseArray = NULL;
879
}
880
881
if (gDatabaseList) {
882
gDatabaseList->clear();
883
gDatabaseList->release();
884
gDatabaseList = NULL;
885
}
886
887
MSHTTPHeaderTable::releaseDefaultMetaDataHeaders();
888
PBMSSystemTables::systemTableShutDown();
889
}
890
891
void MSDatabase::setBackupDatabase()
892
{
893
enter_();
894
// I need to give the backup database a unique fake database ID.
895
// This is so that it is not confused with the database being backed
896
// backed up when opening tables.
897
898
// Normally database IDs are generated by time(NULL) so small database IDs
899
// are safe to use as fake IDs.
900
901
lock_(gDatabaseList);
902
myDatabaseID = PBMSParameters::getBackupDatabaseID() +1;
903
PBMSParameters::setBackupDatabaseID(myDatabaseID);
904
gDatabaseArray->set(myDatabaseID, RETAIN(this));
905
isBackup = true;
906
907
// Notify the cloud storage, if any, that it is a backup.
908
// This is important because if the backup database is dropped
909
// we need to be sure that only the BLOBs belonging to the
910
// backup are removed from the cloud.
911
myBlobCloud->cl_setCloudIsBackup();
912
913
unlock_(gDatabaseList);
914
915
// Rename the database path so that it is obviouse that this is an incomplete backup database.
916
// When the backup is completed it will be renamed back.
917
CSPath *new_path = CSPath::newPath(myDatabasePath->concat("#"));
918
push_(new_path);
919
920
if (new_path->exists())
921
new_path->remove();
922
923
CSPath *db_path = CSPath::newPath(RETAIN(myDatabasePath));
924
push_(db_path);
925
926
db_path->rename(new_path->getNameCString());
927
myDatabasePath->release();
928
myDatabasePath = new_path->getString();
929
myDatabasePath->retain();
930
931
release_(db_path);
932
release_(new_path);
933
934
935
exit_();
936
}
937
938
void MSDatabase::releaseBackupDatabase()
939
{
940
enter_();
941
942
943
// The backup has completed succefully, rename the path to the correct name.
944
CSPath *db_path = CSPath::newPath(myDatabasePath->getCString());
945
push_(db_path);
946
947
myDatabasePath->setLength(myDatabasePath->length()-1);
948
db_path->rename(cs_last_name_of_path(myDatabasePath->getCString()));
949
release_(db_path);
950
951
// Remove the backup database object.
952
lock_(gDatabaseList);
953
gDatabaseArray->remove(myDatabaseID);
954
MSTableList::removeDatabaseTables(this); // Will also release the database object.
955
unlock_(gDatabaseList);
956
957
958
exit_();
959
}
960
961
void MSDatabase::startBackup(MSBackupInfo *backup_info)
962
{
963
enter_();
964
965
push_(backup_info);
966
if (iBackupThread) {
967
if (iBackupThread->isRunning()) {
968
CSException::throwException(CS_CONTEXT, MS_ERR_DUPLICATE_DB, "A backup is still running.");
969
}
970
iBackupThread->release();
971
iBackupThread = NULL;
972
}
973
974
pop_(backup_info);
975
iBackupThread = MSBackup::newMSBackup(backup_info);
976
977
push_(iBackupThread);
978
try_(a) {
979
iBackupThread->startBackup(RETAIN(this));
980
}
981
982
catch_(a) {
983
iBackupThread->release();
984
iBackupThread = NULL;
985
throw_();
986
}
987
cont_(a);
988
989
exit_();
990
}
991
992
bool MSDatabase::backupStatus(uint64_t *total, uint64_t *completed, bool *completed_ok)
993
{
994
bool done;
995
996
enter_();
997
998
if (iBackupThread) {
999
*total = iBackupThread->getBackupSize();
1000
*completed = iBackupThread->getBackupCompletedSize();
1001
done = !iBackupThread->isRunning();
1002
*completed = (iBackupThread->getStatus() == 0);
1003
} else {
1004
*completed_ok = done = true;
1005
*total = *completed = 0;
1006
}
1007
1008
return_(done);
1009
}
1010
1011
uint32_t MSDatabase::backupID()
1012
{
1013
return (iBackupThread)?iBackupThread->backupID(): 0;
1014
}
1015
1016
void MSDatabase::terminateBackup()
1017
{
1018
if (iBackupThread) {
1019
iBackupThread->stop();
1020
iBackupThread->release();
1021
iBackupThread = NULL;
1022
}
1023
}
1024
1025
MSDatabase *MSDatabase::getDatabase(CSString *db_name, bool create)
1026
{
1027
MSDatabase *db;
1028
enter_();
1029
push_(db_name);
1030
1031
1032
lock_(gDatabaseList);
1033
if (!(db = (MSDatabase *) gDatabaseList->find(db_name))) {
1034
db = MSDatabase::loadDatabase(RETAIN(db_name), create);
1035
if (!db)
1036
goto exit;
1037
} else
1038
db->retain();
1039
1040
exit:
1041
unlock_(gDatabaseList);
1042
release_(db_name);
1043
return_(db);
1044
}
1045
1046
MSDatabase *MSDatabase::getDatabase(const char *db_name, bool create)
1047
{
1048
return getDatabase(CSString::newString(db_name), create);
1049
}
1050
1051
MSDatabase *MSDatabase::getDatabase(uint32_t db_id)
1052
{
1053
MSDatabase *db;
1054
1055
enter_();
1056
lock_(gDatabaseList);
1057
if ((db = (MSDatabase *) gDatabaseArray->get((uint32_t) db_id)))
1058
db->retain();
1059
else {
1060
// Look for the database folder with the correct ID:
1061
CSPath *path = CSPath::newPath(PBMSDaemon::getPBMSDir());
1062
push_(path);
1063
if (path->exists()) {
1064
CSDirectory *dir;
1065
dir = CSDirectory::newDirectory(RETAIN(path));
1066
push_(dir);
1067
dir->open();
1068
1069
while (dir->next() && !db) {
1070
if (!dir->isFile()) {
1071
const char *ptr, *dir_name = dir->name();
1072
ptr = dir_name + strlen(dir_name) -1;
1073
1074
while (ptr > dir_name && *ptr != '-') ptr--;
1075
1076
if (*ptr == '-') {
1077
int len = ptr - dir_name;
1078
ptr++;
1079
if ((strtoul(ptr, NULL, 10) == db_id) && len) {
1080
db = getDatabase(CSCString::newString(dir_name, len), true);
1081
ASSERT(db->myDatabaseID == db_id);
1082
}
1083
}
1084
}
1085
}
1086
release_(dir);
1087
}
1088
release_(path);
1089
}
1090
unlock_(gDatabaseList);
1091
1092
if (!db) {
1093
char buffer[CS_EXC_MESSAGE_SIZE];
1094
1095
cs_strcpy(CS_EXC_MESSAGE_SIZE, buffer, "Unknown database #");
1096
cs_strcat(CS_EXC_MESSAGE_SIZE, buffer, (uint32_t) db_id);
1097
CSException::throwException(CS_CONTEXT, MS_ERR_UNKNOWN_DB, buffer);
1098
}
1099
return_(db);
1100
}
1101
1102
void MSDatabase::wakeTempLogThreads()
1103
{
1104
MSDatabase *db;
1105
1106
if (!gDatabaseList)
1107
return;
1108
enter_();
1109
lock_(gDatabaseList);
1110
for (int i=0;;i++) {
1111
if (!(db = (MSDatabase *) gDatabaseList->itemAt(i)))
1112
break;
1113
if (db->myTempLogThread)
1114
db->myTempLogThread->wakeup();
1115
}
1116
unlock_(gDatabaseList);
1117
exit_();
1118
}
1119
1120
uint32_t MSDatabase::getDBID(CSPath *path, CSString *db_name)
1121
{
1122
CSDirectory *dir;
1123
uint32_t db_id = 0;
1124
int len = db_name->length();
1125
const char *ptr;
1126
1127
enter_();
1128
push_(db_name);
1129
push_(path);
1130
1131
// Search for the ID of the database
1132
dir = CSDirectory::newDirectory(RETAIN(path));
1133
push_(dir);
1134
dir->open();
1135
while (dir->next() && !db_id)
1136
{
1137
if (!dir->isFile()){
1138
ptr = dir->name() + strlen(dir->name()) -1;
1139
while (ptr > dir->name() && isdigit(*ptr)) ptr--;
1140
if ((*ptr == '-') && (len == (ptr - dir->name())) && !db_name->compare(dir->name(), len) ) {
1141
db_id = atol(ptr+1);
1142
}
1143
}
1144
}
1145
release_(dir);
1146
1147
if (!db_id) {
1148
db_id = time(NULL);
1149
1150
while (1) { // search for a unique db_id
1151
dir = CSDirectory::newDirectory(RETAIN(path));
1152
push_(dir);
1153
dir->open();
1154
while (db_id && dir->next()) {
1155
if (!dir->isFile()) {
1156
ptr = dir->name() + strlen(dir->name()) -1;
1157
while (ptr > dir->name() && isdigit(*ptr)) ptr--;
1158
if ((*ptr == '-') && (db_id == strtoul(ptr+1, NULL, 10))) {
1159
db_id = 0;
1160
}
1161
}
1162
}
1163
release_(dir);
1164
if (db_id)
1165
break;
1166
sleep(1); // Allow 1 second to pass.
1167
db_id = time(NULL);
1168
}
1169
}
1170
1171
release_(path);
1172
release_(db_name);
1173
return_(db_id);
1174
}
1175
1176
CSPath *MSDatabase::createDatabasePath(const char *location, CSString *db_name, uint32_t *db_id_ptr, bool *create, bool is_pbms)
1177
{
1178
bool create_path = *create;
1179
CSPath *path = NULL;
1180
char name_buffer[MS_DATABASE_NAME_SIZE + 40];
1181
uint32_t db_id;
1182
enter_();
1183
1184
push_(db_name);
1185
*create = false;
1186
path = CSPath::newPath(location, "pbms");
1187
push_(path);
1188
if (!path->exists()) {
1189
if (!create_path){
1190
release_(path);
1191
path = NULL;
1192
goto done;
1193
}
1194
1195
*create = true;
1196
path->makeDir();
1197
}
1198
1199
// If this is the pbms database then nothing more is to be done.
1200
if (is_pbms)
1201
goto done;
1202
1203
if ((!db_id_ptr) || !*db_id_ptr) {
1204
db_id = getDBID(RETAIN(path), RETAIN(db_name));
1205
if (db_id_ptr)
1206
*db_id_ptr = db_id;
1207
} else
1208
db_id = *db_id_ptr;
1209
1210
// Create the PBMS database name with ID
1211
cs_strcpy(MS_DATABASE_NAME_SIZE + 40, name_buffer, db_name->getCString());
1212
cs_strcat(MS_DATABASE_NAME_SIZE + 40, name_buffer, "-");
1213
cs_strcat(MS_DATABASE_NAME_SIZE + 40, name_buffer, (uint32_t) db_id);
1214
1215
pop_(path);
1216
path = CSPath::newPath(path, name_buffer);
1217
push_(path);
1218
if (!path->exists()) {
1219
if (create_path) {
1220
*create = true;
1221
path->makeDir();
1222
} else {
1223
release_(path);
1224
path = NULL;
1225
}
1226
1227
}
1228
1229
done:
1230
if (path)
1231
pop_(path);
1232
release_(db_name);
1233
return_(path);
1234
}
1235
1236
1237
1238
MSDatabase *MSDatabase::newDatabase(const char *db_location, CSString *db_name, uint32_t db_id, bool create)
1239
{
1240
MSDatabase *db = NULL;
1241
CSDirectory *dir;
1242
MSRepository *repo;
1243
CSPath *path;
1244
const char *file_name;
1245
uint32_t file_id;
1246
off64_t file_size;
1247
MSTempLog *log;
1248
uint32_t to_delete = 0;
1249
CSString *db_path;
1250
bool is_pbms = false;
1251
1252
enter_();
1253
1254
push_(db_name);
1255
1256
//is_pbms = (strcmp(db_name->getCString(), "pbms") == 0); To be done later.
1257
1258
/* Block the creation of the pbms database if there is no MySQL database. */
1259
path = CSPath::newPath(ms_my_get_mysql_home_path(), RETAIN(db_name));
1260
push_(path);
1261
if (create && !path->exists()) {
1262
CSException::throwException(CS_CONTEXT, MS_ERR_UNKNOWN_DB, db_name->getCString());
1263
}
1264
release_(path);
1265
1266
// Create the database path, if 'create' == false then it can return NULL
1267
path = createDatabasePath(db_location, RETAIN(db_name), &db_id, &create, is_pbms);
1268
if (!path) {
1269
release_(db_name);
1270
return_(NULL);
1271
}
1272
push_(path);
1273
1274
// Create the database object and initialize it.
1275
if (!(db = new MSDatabase())) {
1276
CSException::throwOSError(CS_CONTEXT, ENOMEM);
1277
}
1278
1279
db->myIsPBMS = is_pbms;
1280
db_path = path->getString();
1281
db_path->retain();
1282
release_(path);
1283
path = NULL;
1284
1285
db->iNextBlobRefId = (uint32_t) time(NULL);
1286
db->iNextBlobRefId <<= 32;
1287
db->iNextBlobRefId = COMMIT_MASK(db->iNextBlobRefId);
1288
db->iNextBlobRefId++;
1289
1290
db->myDatabaseID = db_id;
1291
db->myDatabasePath = db_path;
1292
db->myDatabaseName = db_name;
1293
new_(db->myBlobCloud, CloudDB(db_id));
1294
1295
pop_(db_name);
1296
1297
push_(db);
1298
1299
1300
new_(db->myTempLogArray, CSSyncSparseArray(20));
1301
new_(db->iTableList, CSSyncSortedList());
1302
new_(db->iTableArray, CSSparseArray(20));
1303
new_(db->myRepostoryList, CSSyncVector(20));
1304
1305
if (!is_pbms) { //
1306
#ifdef HAVE_ALIAS_SUPPORT
1307
//db->retain(); no retain here, MSAlias() takes a back ref.
1308
new_(db->iBlobAliases, MSAlias(db));
1309
#endif
1310
/* "Load" the database: */
1311
1312
/* Get the max table ID: */
1313
dir = CSDirectory::newDirectory(RETAIN(db_path));
1314
push_(dir);
1315
dir->open();
1316
while (dir->next()) {
1317
file_name = dir->name();
1318
if (dir->isFile() && cs_is_extension(file_name, "bst"))
1319
db->addTableFromFile(dir, file_name, false);
1320
}
1321
release_(dir);
1322
1323
path = CSPath::newPath(RETAIN(db_path), "bs-repository");
1324
if (path->exists()) {
1325
dir = CSDirectory::newDirectory(path);
1326
push_(dir);
1327
dir->open();
1328
while (dir->next()) {
1329
file_name = dir->name();
1330
if (dir->isFile() && cs_is_extension(file_name, "bs")) {
1331
if ((file_id = fileToTableId(file_name, "repo"))) {
1332
dir->info(NULL, &file_size, NULL);
1333
new_(repo, MSRepository(file_id, db, file_size));
1334
db->myRepostoryList->set(file_id - 1, repo);
1335
}
1336
}
1337
}
1338
release_(dir);
1339
}
1340
else {
1341
path->makeDir();
1342
path->release();
1343
}
1344
1345
path = CSPath::newPath(RETAIN(db_path), "bs-logs");
1346
if (path->exists()) {
1347
dir = CSDirectory::newDirectory(path);
1348
push_(dir);
1349
dir->open();
1350
while (dir->next()) {
1351
file_name = dir->name();
1352
if (dir->isFile()) {
1353
if (cs_is_extension(file_name, "bs")) {
1354
if ((file_id = fileToTableId(file_name, "temp"))) {
1355
dir->info(NULL, &file_size, NULL);
1356
new_(log, MSTempLog(file_id, db, file_size));
1357
db->myTempLogArray->set(file_id, log);
1358
}
1359
}
1360
else if (cs_is_extension(file_name, "bst")) {
1361
db->addTableFromFile(dir, file_name, true);
1362
to_delete++;
1363
}
1364
}
1365
}
1366
release_(dir);
1367
}
1368
else {
1369
path->makeDir();
1370
path->release();
1371
}
1372
1373
if (to_delete) {
1374
/* Go through and prepare all the tables that are to
1375
* be deleted:
1376
*/
1377
uint32_t i = 0;
1378
MSTable *tab;
1379
1380
while ((tab = (MSTable *) db->iTableList->itemAt(i))) {
1381
if (tab->isToDelete())
1382
tab->prepareToDelete();
1383
i++;
1384
}
1385
}
1386
1387
}
1388
pop_(db);
1389
1390
return_(db);
1391
}
1392
1393
void MSDatabase::startThreads()
1394
{
1395
enter_();
1396
1397
if (myIsPBMS)
1398
exit_();
1399
1400
#ifdef HAVE_ALIAS_SUPPORT
1401
// iBlobAliases->ma_open() must be called before starting any threads.
1402
iBlobAliases->ma_open();
1403
#endif
1404
1405
new_(myTempLogThread, MSTempLogThread(1 * 1000, this));
1406
myTempLogThread->start();
1407
1408
#ifdef MS_COMPACTOR_POLLS
1409
new_(myCompactorThread, MSCompactorThread(MS_COMPACTOR_POLL_FREQ, this));
1410
#else
1411
new_(myCompactorThread, MSCompactorThread(MS_DEFAULT_COMPACTOR_WAIT * 1000, this));
1412
#endif
1413
1414
1415
myCompactorThread->start();
1416
exit_();
1417
}
1418
1419
1420
void MSDatabase::dropDatabase()
1421
{
1422
enter_();
1423
1424
iDropping = true;
1425
iClosing = true;
1426
1427
if (iBackupThread) {
1428
iBackupThread->stop();
1429
iBackupThread->release();
1430
iBackupThread = NULL;
1431
}
1432
1433
if (myTempLogThread) {
1434
myTempLogThread->stop();
1435
myTempLogThread->release();
1436
myTempLogThread = NULL;
1437
}
1438
1439
if (myCompactorThread) {
1440
myRepostoryList->wakeup(); // The compator thread waits on this.
1441
myCompactorThread->stop();
1442
myCompactorThread->release();
1443
myCompactorThread = NULL;
1444
}
1445
1446
// Call cloud drop database even if the database is not currrently
1447
// using cloud storage just in case to was in the past. If the connection
1448
// to the cloud is not setup then nothing will be done.
1449
try_(a) {
1450
myBlobCloud->cl_dropDB();
1451
}
1452
catch_(a) {
1453
self->logException();
1454
}
1455
cont_(a);
1456
exit_();
1457
}
1458
1459
void MSDatabase::removeDatabasePath(CSString *doomedDatabasePath )
1460
{
1461
CSPath *path = NULL;
1462
CSDirectory *dir = NULL;
1463
const char *file_name;
1464
enter_();
1465
1466
push_(doomedDatabasePath);
1467
1468
// Delete repository files
1469
path = CSPath::newPath(RETAIN(doomedDatabasePath), "bs-repository");
1470
push_(path);
1471
if (path->exists()) {
1472
dir = CSDirectory::newDirectory(RETAIN(path));
1473
push_(dir);
1474
dir->open();
1475
while (dir->next()) {
1476
file_name = dir->name();
1477
if (dir->isFile() && cs_is_extension(file_name, "bs")) {
1478
dir->deleteEntry();
1479
}
1480
}
1481
release_(dir);
1482
if (path->isEmpty())
1483
path->removeDir();
1484
}
1485
release_(path);
1486
1487
// Delete temp log files.
1488
path = CSPath::newPath(RETAIN(doomedDatabasePath), "bs-logs");
1489
push_(path);
1490
if (path->exists()) {
1491
dir = CSDirectory::newDirectory(RETAIN(path));
1492
push_(dir);
1493
dir->open();
1494
while (dir->next()) {
1495
file_name = dir->name();
1496
if (dir->isFile() && (cs_is_extension(file_name, "bs") || cs_is_extension(file_name, "bst"))) {
1497
dir->deleteEntry();
1498
}
1499
}
1500
release_(dir);
1501
if (path->isEmpty())
1502
path->removeDir();
1503
}
1504
release_(path);
1505
1506
// Delete table reference files.
1507
dir = CSDirectory::newDirectory(RETAIN(doomedDatabasePath));
1508
push_(dir);
1509
dir->open();
1510
while (dir->next()) {
1511
file_name = dir->name();
1512
if (dir->isFile() && cs_is_extension(file_name, "bst"))
1513
dir->deleteEntry();
1514
}
1515
release_(dir);
1516
1517
#ifdef HAVE_ALIAS_SUPPORT
1518
path = CSPath::newPath(RETAIN(doomedDatabasePath), ACTIVE_ALIAS_INDEX);
1519
push_(path);
1520
path->removeFile();
1521
release_(path);
1522
#endif
1523
1524
PBMSSystemTables::removeSystemTables(RETAIN(doomedDatabasePath));
1525
1526
path = CSPath::newPath(RETAIN(doomedDatabasePath));
1527
push_(path);
1528
if (path->isEmpty() && !path->isLink()) {
1529
path->removeDir();
1530
} else {
1531
CSStringBuffer *new_name;
1532
// If the database folder is not empty we rename it to get it out of the way.
1533
// If it is not renamed it will be reused if a database with the same name is
1534
// created again wich will result in the database ID being reused which may
1535
// have some bad side effects.
1536
new_(new_name, CSStringBuffer());
1537
push_(new_name);
1538
new_name->append(cs_last_name_of_path(doomedDatabasePath->getCString()));
1539
new_name->append("_DROPPED");
1540
path->rename(new_name->getCString());
1541
release_(new_name);
1542
}
1543
release_(path);
1544
1545
release_(doomedDatabasePath);
1546
1547
path = CSPath::newPath(PBMSDaemon::getPBMSDir());
1548
push_(path);
1549
if (path->isEmpty() && !path->isLink()) {
1550
path->removeDir();
1551
}
1552
release_(path);
1553
1554
exit_();
1555
}
1556
1557
/* Drop the PBMS database if it exists.
1558
* The root folder 'pbms' will be deleted also
1559
* if it is empty and not a symbolic link.
1560
* The database folder in 'pbms' is deleted if it is empty and
1561
* it is not a symbolic link.
1562
*/
1563
void MSDatabase::dropDatabase(MSDatabase *doomedDatabase, const char *db_name )
1564
{
1565
CSString *doomedDatabasePath = NULL;
1566
1567
enter_();
1568
1569
if (doomedDatabase) {
1570
push_(doomedDatabase);
1571
1572
// Remove any pending transactions for the dropped database.
1573
// This is important because if the database is restored it will have the
1574
// same database ID and the old transactions would be applied to it.
1575
MSTransactionManager::dropDatabase(doomedDatabase->myDatabaseID);
1576
1577
doomedDatabasePath = doomedDatabase->myDatabasePath;
1578
doomedDatabasePath->retain(); // Hold on to this path after the database has been released.
1579
1580
MSTableList::removeDatabaseTables(RETAIN(doomedDatabase));
1581
MSSystemTableShare::removeDatabaseSystemTables(RETAIN(doomedDatabase));
1582
1583
doomedDatabase->dropDatabase(); // Shutdown database threads.
1584
1585
// To avoid a deadlock a lock is not taken on the database list
1586
// if shutdown is in progress. The only database that would be
1587
// dropped during a shutdown is an incomplete backup database.
1588
ASSERT(doomedDatabase->isBackup || !self->myMustQuit);
1589
if (!self->myMustQuit)
1590
lock_(gDatabaseList); // Be sure to shutdown the database before locking this or it can lead to deadlocks
1591
1592
gDatabaseArray->remove(doomedDatabase->myDatabaseID);
1593
if (!doomedDatabase->isBackup)
1594
gDatabaseList->remove(doomedDatabase->getKey());
1595
if (!self->myMustQuit)
1596
unlock_(gDatabaseList);
1597
ASSERT(doomedDatabase->iRefCount == 1);
1598
release_(doomedDatabase);
1599
1600
} else {
1601
CSPath *path;
1602
bool create = false;
1603
uint32_t db_id;
1604
1605
path = createDatabasePath(ms_my_get_mysql_home_path(), CSString::newString(db_name), &db_id, &create);
1606
1607
if (path) {
1608
MSTransactionManager::dropDatabase(db_id);
1609
1610
push_(path);
1611
doomedDatabasePath = path->getString();
1612
doomedDatabasePath->retain(); // Hold on to this path after the database has been released.
1613
release_(path);
1614
}
1615
}
1616
1617
if (doomedDatabasePath)
1618
removeDatabasePath(doomedDatabasePath);
1619
1620
exit_();
1621
}
1622
1623
void MSDatabase::dropDatabase(const char *db_name )
1624
{
1625
enter_();
1626
dropDatabase(getDatabase(db_name, false), db_name);
1627
exit_();
1628
}
1629
1630
// The table_path can be several things here:
1631
// 1: <absalute path>/<database>/<table>
1632
// 2: <absalute path>/<database>
1633
// 3: <database>/<table>
1634
bool MSDatabase::convertTablePathToIDs(const char *table_path, uint32_t *db_id, uint32_t *tab_id, bool create)
1635
{
1636
const char *base = ms_my_get_mysql_home_path();
1637
CSString *table_url;
1638
CSString *db_path = NULL;
1639
CSString *db_name = NULL;
1640
CSString *tab_name = NULL;
1641
MSDatabase *db;
1642
enter_();
1643
1644
*db_id = 0;
1645
*tab_id = 0;
1646
1647
table_url = CSString::newString(table_path);
1648
if (table_url->startsWith(base)) {
1649
table_url = table_url->right(base);
1650
}
1651
push_(table_url);
1652
1653
1654
db_path = table_url->left("/", -1);
1655
push_(db_path);
1656
tab_name = table_url->right("/", -1);
1657
1658
pop_(db_path);
1659
release_(table_url);
1660
1661
if (db_path->length() == 0) { // Only a database name was supplied.
1662
db_path->release();
1663
db_name = tab_name;
1664
tab_name = NULL;
1665
} else {
1666
if (tab_name->length() == 0) {
1667
tab_name->release();
1668
tab_name = NULL;
1669
} else
1670
push_(tab_name);
1671
push_(db_path);
1672
db_name = db_path->right("/", -1);
1673
pop_(db_path);
1674
if (db_name->length() == 0) {
1675
db_name->release();
1676
db_name = db_path;
1677
} else {
1678
db_path->release();
1679
db_path = NULL;
1680
}
1681
}
1682
1683
db = MSDatabase::getDatabase(db_name, create); // This will release db_name
1684
if (db) {
1685
*db_id = db->myDatabaseID;
1686
if (tab_name) {
1687
MSTable *tab;
1688
pop_(tab_name);
1689
push_(db);
1690
tab = db->getTable(tab_name, create);// This will release tab_name
1691
pop_(db);
1692
if (tab) {
1693
*tab_id = tab->myTableID;
1694
tab->release();
1695
}
1696
}
1697
1698
db->release();
1699
}
1700
1701
return_((*tab_id > 0) && (*db_id > 0));
1702
}
1703
1704
bool MSDatabase::convertTableAndDatabaseToIDs(const char *db_name, const char *tab_name, uint32_t *db_id, uint32_t *tab_id, bool create)
1705
{
1706
MSDatabase *db;
1707
enter_();
1708
1709
*db_id = 0;
1710
*tab_id = 0;
1711
1712
db = MSDatabase::getDatabase(db_name, create);
1713
if (db) {
1714
push_(db);
1715
*db_id = db->myDatabaseID;
1716
if (tab_name) {
1717
MSTable *tab;
1718
tab = db->getTable(tab_name, create);
1719
if (tab) {
1720
*tab_id = tab->myTableID;
1721
tab->release();
1722
}
1723
}
1724
1725
release_(db);
1726
}
1727
1728
return_((*tab_id > 0) && (*db_id > 0));
1729
}
1730
1731
MSDatabase *MSDatabase::loadDatabase(CSString *db_name, bool create)
1732
{
1733
MSDatabase *db;
1734
enter_();
1735
1736
db = newDatabase(ms_my_get_mysql_home_path(), db_name, 0, create);
1737
1738
if (db) {
1739
push_(db);
1740
1741
gDatabaseList->add(RETAIN(db));
1742
1743
gDatabaseArray->set(db->myDatabaseID, RETAIN(db));
1744
db->startThreads();
1745
PBMSSystemTables::loadSystemTables(RETAIN(db));
1746
1747
pop_(db);
1748
}
1749
return_(db);
1750
}
1751
1752
uint32_t MSDatabase::getDatabaseID(CSString *db_name, bool create)
1753
{
1754
MSDatabase *db;
1755
uint32_t id = 0;
1756
enter_();
1757
push_(db_name);
1758
1759
1760
lock_(gDatabaseList);
1761
if (!(db = (MSDatabase *) gDatabaseList->find(db_name))) {
1762
db = MSDatabase::loadDatabase(RETAIN(db_name), create);
1763
if (!db)
1764
goto exit;
1765
id = db->myDatabaseID;
1766
db->release();
1767
} else
1768
id = db->myDatabaseID;
1769
1770
exit:
1771
unlock_(gDatabaseList);
1772
release_(db_name);
1773
return_(id);
1774
}
1775
1776
1777
uint32_t MSDatabase::getDatabaseID(const char *db_name, bool create)
1778
{
1779
return getDatabaseID(CSString::newString(db_name), create);
1780
}
1781
1782
MSDatabase *MSDatabase::getBackupDatabase(CSString *db_location, CSString *db_name, uint32_t db_id, bool create)
1783
{
1784
bool was_created = create;
1785
CSPath *path;
1786
MSDatabase *db;
1787
enter_();
1788
1789
push_(db_location);
1790
push_(db_name);
1791
// If the db already exists and 'create' == true then the existing db
1792
// must be deleted.
1793
// Create the database path, if 'create' == false then it can return NULL
1794
path = createDatabasePath(db_location->getCString(), RETAIN(db_name), &db_id, &was_created);
1795
if (!path) {
1796
CSException::throwException(CS_CONTEXT, MS_ERR_UNKNOWN_DB, db_name->getCString());
1797
}
1798
push_(path);
1799
1800
// If we wanted to create it but it already exists then throw an error.
1801
if ( create && !was_created) {
1802
char str[120];
1803
snprintf(str, 120, "Duplicate database: %s", db_name->getCString());
1804
CSException::throwException(CS_CONTEXT, MS_ERR_DUPLICATE_DB, str);
1805
}
1806
1807
release_(path);
1808
pop_(db_name);
1809
// everything looks OK
1810
db = newDatabase(db_location->getCString(), db_name, db_id, create);
1811
db->setBackupDatabase();
1812
release_(db_location);
1813
return_(db);
1814
}
1815
Loggerhead is a web-based interface for Breezy
Version: 2.0.1