1
/* Copyright (C) 2008 PrimeBase Technologies GmbH, Germany
3
* PrimeBase Media Stream for MySQL
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.
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.
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., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
19
* Original author: Paul McCullagh
20
* Continued development: Barry Leslie
26
* Media Stream Tables.
29
#include "cslib/CSConfig.h"
33
#include "cslib/CSGlobal.h"
34
#include "cslib/CSLog.h"
35
#include "cslib/CSStrUtil.h"
36
#include "cslib/CSPath.h"
38
#include "open_table_ms.h"
40
#include "connection_handler_ms.h"
41
#include "engine_ms.h"
42
#include "transaction_ms.h"
43
#include "parameters_ms.h"
46
* ---------------------------------------------------------------
50
MSOpenTable::MSOpenTable():
59
myWriteRepoFile(NULL),
67
memset(myOTBuffer, 0, MS_OT_BUFFER_SIZE); // wipe this to make valgrind happy.
70
MSOpenTable::~MSOpenTable()
75
void MSOpenTable::close()
79
myTableFile->release();
84
myTempLogFile->release();
98
void MSOpenTable::returnToPool()
100
MSTableList::releaseTable(this);
103
// This cleanup class is used to reset the
104
// repository size if something goes wrong.
105
class CreateBlobCleanUp : public CSRefObject {
113
CreateBlobCleanUp(): CSRefObject(),
119
repo->setRepoFileSize(ot, old_size);
124
void setCleanUp(MSOpenTable *ot_arg, MSRepository *repo_arg, uint64_t size)
139
void MSOpenTable::createBlob(PBMSBlobURLPtr bh, uint64_t blob_size, char *metadata, uint16_t metadata_size, CSInputStream *stream, CloudKeyPtr cloud_key, Md5Digest *checksum)
141
uint64_t repo_offset;
142
uint64_t blob_id = 0;
150
Md5Digest my_checksum;
151
CloudKeyRec cloud_key_rec;
152
CreateBlobCleanUp *cleanup;
155
new_(cleanup, CreateBlobCleanUp());
159
checksum = &my_checksum;
161
if (stream) push_(stream);
164
auth_code = random();
165
repo_size = myWriteRepo->getRepoFileSize();
166
temp_time = myWriteRepo->myLastTempTime;
168
// If an exception occurs the cleanup operation will be called.
169
cleanup->setCleanUp(this, myWriteRepo, repo_size);
171
head_size = myWriteRepo->getDefaultHeaderSize(metadata_size);
172
if (getDB()->myBlobType == MS_STANDARD_STORAGE) {
174
repo_offset = myWriteRepo->receiveBlob(this, head_size, blob_size, checksum, stream);
176
ASSERT(getDB()->myBlobType == MS_CLOUD_STORAGE);
177
CloudDB *cloud = getDB()->myBlobCloud;
180
CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "Creating cloud BLOB without cloud.");
182
repo_offset = repo_size + head_size;
183
memset(checksum, 0, sizeof(Md5Digest)); // The checksum is only for local storage.
185
// If there is a stream then the data has not been sent to the cloud yet.
187
cloud_key = &cloud_key_rec;
188
cloud->cl_getNewKey(cloud_key);
190
cloud->cl_putData(cloud_key, stream, blob_size);
195
repo_id = myWriteRepo->myRepoID;
197
getDB()->queueForDeletion(this, MS_TL_REPO_REF, repo_id, repo_offset, auth_code, &log_id, &log_offset, &temp_time);
198
formatRepoURL(bh, repo_id, repo_offset, auth_code, blob_size);
201
blob_id = getDBTable()->createBlobHandle(this, myWriteRepo->myRepoID, repo_offset, blob_size, head_size, auth_code);
202
getDB()->queueForDeletion(this, MS_TL_BLOB_REF, getDBTable()->myTableID, blob_id, auth_code, &log_id, &log_offset, &temp_time);
203
formatBlobURL(bh, blob_id, auth_code, blob_size, 0);
206
myWriteRepo->writeBlobHead(this, repo_offset, myWriteRepo->myRepoDefRefSize, head_size, blob_size, checksum, metadata, metadata_size, blob_id, auth_code, log_id, log_offset, getDB()->myBlobType, cloud_key);
208
cleanup->cancelCleanUp();
214
// BLOBs created with this method are always created as standard local BLOBs. (No cloud storage)
215
void MSOpenTable::createBlob(PBMSBlobIDPtr blob_id, uint64_t blob_size, char *metadata, uint16_t metadata_size)
218
uint64_t repo_offset;
225
CreateBlobCleanUp *cleanup;
228
new_(cleanup, CreateBlobCleanUp());
233
auth_code = random();
235
repo_size = myWriteRepo->getRepoFileSize();
237
// If an exception occurs the cleanup operation will be called.
238
cleanup->setCleanUp(this, myWriteRepo, repo_size);
240
head_size = myWriteRepo->getDefaultHeaderSize(metadata_size);
242
repo_offset = myWriteRepo->receiveBlob(this, head_size, blob_size);
243
repo_id = myWriteRepo->myRepoID;
244
temp_time = myWriteRepo->myLastTempTime;
245
getDB()->queueForDeletion(this, MS_TL_REPO_REF, repo_id, repo_offset, auth_code, &log_id, &log_offset, &temp_time);
246
myWriteRepo->myLastTempTime = temp_time;
247
myWriteRepo->writeBlobHead(this, repo_offset, myWriteRepo->myRepoDefRefSize, head_size, blob_size, NULL, metadata, metadata_size, 0, auth_code, log_id, log_offset, MS_STANDARD_STORAGE, NULL);
248
// myWriteRepo->setRepoFileSize(this, repo_offset + head_size + blob_size);This is now set by writeBlobHead()
250
blob_id->bi_db_id = getDB()->myDatabaseID;
251
blob_id->bi_blob_id = repo_offset;
252
blob_id->bi_tab_id = repo_id;
253
blob_id->bi_auth_code = auth_code;
254
blob_id->bi_blob_size = blob_size;
255
blob_id->bi_blob_type = MS_URL_TYPE_REPO;
256
blob_id->bi_blob_ref_id = 0;
258
cleanup->cancelCleanUp();
264
void MSOpenTable::sendRepoBlob(uint64_t blob_id, uint64_t req_offset, uint64_t req_size, uint32_t auth_code, bool info_only, CSHTTPOutputStream *stream)
270
MSRepoFile *repo_file;
274
getDBTable()->readBlobHandle(this, blob_id, &auth_code, &repo_id, &offset, &size, &head_size, true);
275
repo_file = getDB()->getRepoFileFromPool(repo_id, false);
276
frompool_(repo_file);
277
//repo_file->sendBlob(this, offset, head_size, size, stream);
278
repo_file->sendBlob(this, offset, req_offset, req_size, 0, false, info_only, stream);
279
backtopool_(repo_file);
283
void MSOpenTable::freeReference(uint64_t blob_id, uint64_t blob_ref_id)
289
MSRepoFile *repo_file;
290
uint32_t auth_code = 0;
295
getDBTable()->readBlobHandle(this, blob_id, &auth_code, &repo_id, &offset, &blob_size, &head_size, true);
296
repo_file = getDB()->getRepoFileFromPool(repo_id, false);
298
frompool_(repo_file);
299
repo_file->releaseBlob(this, offset, head_size, getDBTable()->myTableID, blob_id, blob_ref_id, auth_code);
300
backtopool_(repo_file);
305
void MSOpenTable::commitReference(uint64_t blob_id, uint64_t blob_ref_id)
311
MSRepoFile *repo_file;
312
uint32_t auth_code = 0;
317
getDBTable()->readBlobHandle(this, blob_id, &auth_code, &repo_id, &offset, &blob_size, &head_size, true);
318
repo_file = getDB()->getRepoFileFromPool(repo_id, false);
320
frompool_(repo_file);
321
repo_file->commitBlob(this, offset, head_size, getDBTable()->myTableID, blob_id, blob_ref_id, auth_code);
322
backtopool_(repo_file);
327
void MSOpenTable::useBlob(int type, uint32_t db_id, uint32_t tab_id, uint64_t blob_id, uint32_t auth_code, uint16_t col_index, uint64_t blob_size, uint64_t blob_ref_id, PBMSBlobURLPtr ret_blob_url)
329
MSRepoFile *repo_file= NULL;
331
CSInputStream *stream;
335
uint64_t repo_offset;
342
if (!blob_db->isRecovering()) {
343
// During recovery the only thing that needs to be done is to
344
// reset the database ID which is done when the URL is created.
345
// Create the URL using the table ID passed in not the one from
346
// the table associated with this object.
349
if (type == MS_URL_TYPE_REPO) { // There is no table reference associated with this BLOB yet.
354
if (blob_db->myDatabaseID == db_id)
355
repo_file = blob_db->getRepoFileFromPool(tab_id, false);
358
blob_db = MSDatabase::getDatabase(db_id);
360
repo_file = blob_db->getRepoFileFromPool(tab_id, false);
362
blob_db = repo_file->myRepo->myRepoDatabase;
365
frompool_(repo_file);
366
repo_file->read(&blob, blob_id, MS_MIN_BLOB_HEAD_SIZE, MS_MIN_BLOB_HEAD_SIZE);
368
repo_offset = blob_id;
369
blob_size = CS_GET_DISK_6(blob.rb_blob_data_size_6);
370
head_size = CS_GET_DISK_2(blob.rb_head_size_2);
372
ac = CS_GET_DISK_4(blob.rb_auth_code_4);
374
CSException::throwException(CS_CONTEXT, MS_ERR_AUTH_FAILED, "Invalid BLOB identifier");
375
status = CS_GET_DISK_1(blob.rb_status_1);
376
if ( ! IN_USE_BLOB_STATUS(status))
377
CSException::throwException(CS_CONTEXT, MS_ERR_NOT_FOUND, "BLOB has already been deleted");
380
// Create a table reference to the BLOB:
382
blob_id = getDBTable()->createBlobHandle(this, tab_id, blob_id, blob_size, head_size, auth_code);
383
state = MS_UB_NEW_HANDLE;
387
getDB()->openWriteRepo(this);
389
// If either databases are using cloud storage then this is
390
// not supported yet.
391
if (getDB()->myBlobCloud || myWriteRepo->myRepoDatabase->myBlobCloud)
392
CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "Copying cloud BLOB between databases is not supported.");
394
stream = repo_file->getInputStream(repo_offset);
396
repo_offset = myWriteRepo->copyBlob(this, head_size + blob_size, stream);
399
// Create a table reference to the BLOB:
400
repo_id = myWriteRepo->myRepoID;
401
blob_id = getDBTable()->createBlobHandle(this, myWriteRepo->myRepoID, repo_offset, blob_size, head_size, auth_code);
402
state = MS_UB_NEW_BLOB;
404
backtopool_(repo_file);
408
if (blob_db->myDatabaseID == db_id && getDBTable()->myTableID == tab_id) {
409
getDBTable()->readBlobHandle(this, blob_id, &auth_code, &repo_id, &repo_offset, &blob_size, &head_size, true);
411
state = MS_UB_SAME_TAB;
414
MSOpenTable *blob_otab;
416
blob_otab = MSTableList::getOpenTableByID(db_id, tab_id);
417
frompool_(blob_otab);
418
blob_otab->getDBTable()->readBlobHandle(blob_otab, blob_id, &auth_code, &repo_id, &repo_offset, &blob_size, &head_size, true);
419
if (blob_db->myDatabaseID == db_id) {
420
blob_id = getDBTable()->findBlobHandle(this, repo_id, repo_offset, blob_size, head_size, auth_code);
422
blob_id = getDBTable()->createBlobHandle(this, repo_id, repo_offset, blob_size, head_size, auth_code);
423
state = MS_UB_NEW_HANDLE;
427
// If either databases are using cloud storage then this is
428
// not supported yet.
429
if (blob_db->myBlobCloud || myWriteRepo->myRepoDatabase->myBlobCloud)
430
CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "Copying cloud BLOB between databases is not supported.");
432
// NOTE: For each BLOB reference copied from one database to another a new
433
// BLOB will be created. This can result in multiple copies fo the same BLOB
434
// in the destination database. One way around this would be to redisign things
435
// so that there is one BLOB repository shared across all databases.
436
blob_db->openWriteRepo(this);
438
stream = repo_file->getInputStream(repo_offset);
441
repo_offset = myWriteRepo->copyBlob(this, head_size + blob_size, stream);
445
repo_id = myWriteRepo->myRepoID;
446
blob_id = getDBTable()->createBlobHandle(this, myWriteRepo->myRepoID, repo_offset, blob_size, head_size, auth_code);
447
state = MS_UB_NEW_BLOB;
449
backtopool_(blob_otab);
454
blob_ref_id = blob_db->newBlobRefId();
456
// Always use the table ID of this table because regardless of
457
// where the BLOB ref came from it is being inserted into this table.
458
tab_id = getDBTable()->myTableID;
460
// Add the BLOB reference to the repository.
461
repo_file = blob_db->getRepoFileFromPool(repo_id, false);
462
frompool_(repo_file);
463
repo_file->referenceBlob(this, repo_offset, head_size, tab_id, blob_id, blob_ref_id, auth_code, col_index);
464
backtopool_(repo_file);
466
MSTransactionManager::referenceBLOB(getDB()->myDatabaseID, tab_id, blob_id, blob_ref_id);
470
formatBlobURL(ret_blob_url, blob_id, auth_code, blob_size, tab_id, blob_ref_id);
475
void MSOpenTable::releaseReference(uint64_t blob_id, uint64_t blob_ref_id)
479
MSTransactionManager::dereferenceBLOB(getDB()->myDatabaseID, getDBTable()->myTableID, blob_id, blob_ref_id);
484
void MSOpenTable::checkBlob(CSStringBuffer *buffer, uint64_t blob_id, uint32_t auth_code, uint32_t temp_log_id, uint32_t temp_log_offset)
490
MSRepoFile *repo_file;
494
if (getDBTable()->readBlobHandle(this, blob_id, &auth_code, &repo_id, &offset, &size, &head_size, false)) {
495
if ((repo_file = getDB()->getRepoFileFromPool(repo_id, true))) {
496
frompool_(repo_file);
497
repo_file->checkBlob(buffer, offset, auth_code, temp_log_id, temp_log_offset);
498
backtopool_(repo_file);
501
getDBTable()->freeBlobHandle(this, blob_id, repo_id, offset, auth_code);
506
bool MSOpenTable::deleteReferences(uint32_t temp_log_id, uint32_t temp_log_offset, bool *must_quit)
508
MSTableHeadRec tab_head;
510
MSTableBlobRec tab_blob;
512
uint64_t repo_offset;
515
MSRepoFile *repo_file = NULL;
520
if (myTableFile->read(&tab_head, 0, offsetof(MSTableHeadRec, th_reserved_4), 0) < offsetof(MSTableHeadRec, th_reserved_4))
521
/* Nothing to read, delete it ... */
523
if (CS_GET_DISK_4(tab_head.th_temp_log_id_4) != temp_log_id ||
524
CS_GET_DISK_4(tab_head.th_temp_log_offset_4) != temp_log_offset) {
525
/* Wrong delete reference (ignore): */
530
blob_id = CS_GET_DISK_2(tab_head.th_head_size_2);
531
while (blob_id + sizeof(MSTableBlobRec) <= getDBTable()->getTableFileSize()) {
533
/* Bit of a waste of work, but we must quit! */
537
if (myTableFile->read(&tab_blob, blob_id, sizeof(MSTableBlobRec), 0) < sizeof(MSTableBlobRec))
539
repo_id = CS_GET_DISK_3(tab_blob.tb_repo_id_3);
540
repo_offset = CS_GET_DISK_6(tab_blob.tb_offset_6);
541
head_size = CS_GET_DISK_2(tab_blob.tb_header_size_2);
542
auth_code = CS_GET_DISK_4(tab_blob.tb_auth_code_4);
543
if (repo_file && repo_file->myRepo->myRepoID != repo_id) {
544
backtopool_(repo_file);
548
repo_file = getDB()->getRepoFileFromPool(repo_id, true);
550
frompool_(repo_file);
553
repo_file->freeTableReference(this, repo_offset, head_size, getDBTable()->myTableID, blob_id, auth_code);
555
blob_id += sizeof(MSTableBlobRec);
559
backtopool_(repo_file);
565
void MSOpenTable::openForReading()
567
if (!myTableFile && !isNotATable)
568
myTableFile = getDBTable()->openTableFile();
571
void MSOpenTable::openForWriting()
573
if (myTableFile && myWriteRepo && myWriteRepoFile)
577
if (!myWriteRepo || !myWriteRepoFile)
578
getDB()->openWriteRepo(this);
582
void MSOpenTable::closeForWriting()
584
if (myWriteRepoFile) {
585
myWriteRepoFile->myRepo->syncHead(myWriteRepoFile);
586
myWriteRepoFile->release();
587
myWriteRepoFile = NULL;
590
myWriteRepo->unlockRepo(REPO_WRITE);
591
#ifndef MS_COMPACTOR_POLLS
592
if (myWriteRepo->getGarbageLevel() >= PBMSParameters::getGarbageThreshold()) {
593
if (myWriteRepo->myRepoDatabase->myCompactorThread)
594
myWriteRepo->myRepoDatabase->myCompactorThread->wakeup();
597
myWriteRepo->release();
602
uint32_t MSOpenTable::getTableID()
604
return myPool->myPoolTable->myTableID;
607
MSTable *MSOpenTable::getDBTable()
609
return myPool->myPoolTable;
612
MSDatabase *MSOpenTable::getDB()
614
return myPool->myPoolDB;
617
void MSOpenTable::formatBlobURL(PBMSBlobURLPtr blob_url, uint64_t blob_id, uint32_t auth_code, uint64_t blob_size, uint32_t tab_id, uint64_t blob_ref_id)
621
blob.bu_type = MS_URL_TYPE_BLOB;
622
blob.bu_db_id = getDB()->myDatabaseID;
623
blob.bu_tab_id = tab_id;
624
blob.bu_blob_id = blob_id;
625
blob.bu_auth_code = auth_code;
626
blob.bu_server_id = PBMSParameters::getServerID();
627
blob.bu_blob_size = blob_size;
628
blob.bu_blob_ref_id = blob_ref_id;
630
PBMSBlobURLTools::buildBlobURL(&blob, blob_url);
633
void MSOpenTable::formatBlobURL(PBMSBlobURLPtr blob_url, uint64_t blob_id, uint32_t auth_code, uint64_t blob_size, uint64_t blob_ref_id)
635
formatBlobURL(blob_url, blob_id, auth_code, blob_size, getDBTable()->myTableID, blob_ref_id);
637
void MSOpenTable::formatRepoURL(PBMSBlobURLPtr blob_url, uint32_t log_id, uint64_t log_offset, uint32_t auth_code, uint64_t blob_size)
641
blob.bu_type = MS_URL_TYPE_REPO;
642
blob.bu_db_id = getDB()->myDatabaseID;
643
blob.bu_tab_id = log_id;
644
blob.bu_blob_id = log_offset;
645
blob.bu_auth_code = auth_code;
646
blob.bu_server_id = PBMSParameters::getServerID();
647
blob.bu_blob_size = blob_size;
648
blob.bu_blob_ref_id = 0;
650
PBMSBlobURLTools::buildBlobURL(&blob, blob_url);
653
MSOpenTable *MSOpenTable::newOpenTable(MSOpenTablePool *pool)
657
if (!(otab = new MSOpenTable()))
658
CSException::throwOSError(CS_CONTEXT, ENOMEM);
659
if ((otab->myPool = pool))
660
otab->isNotATable = pool->myPoolTable == NULL;
662
otab->isNotATable = false;
668
* ---------------------------------------------------------------
672
MSOpenTablePool::MSOpenTablePool():
681
MSOpenTablePool::~MSOpenTablePool()
684
removeOpenTablesNotInUse();
685
/* With this, I also delete those that are in use!: */
688
myPoolTable->release();
694
void MSOpenTablePool::check()
696
MSOpenTable *otab, *ptab;
699
if ((otab = (MSOpenTable *) iPoolTables.getBack())) {
708
ptab = ptab->nextTable;
716
otab = (MSOpenTable *) otab->getNextLink();
725
* This returns the table referenced. So it is safe from the pool being
728
MSOpenTable *MSOpenTablePool::getPoolTable()
732
if ((otab = iTablePool)) {
733
iTablePool = otab->nextTable;
734
otab->nextTable = NULL;
735
ASSERT(!otab->inUse);
742
void MSOpenTablePool::returnOpenTable(MSOpenTable *otab)
745
otab->nextTable = iTablePool;
750
* Add a table to the pool, but do not release it!
752
void MSOpenTablePool::addOpenTable(MSOpenTable *otab)
754
iPoolTables.addFront(otab);
757
void MSOpenTablePool::removeOpenTable(MSOpenTable *otab)
760
iPoolTables.remove(otab);
763
void MSOpenTablePool::removeOpenTablesNotInUse()
765
MSOpenTable *otab, *curr_otab;
768
/* Remove all tables that are not in use: */
769
if ((otab = (MSOpenTable *) iPoolTables.getBack())) {
772
otab = (MSOpenTable *) otab->getNextLink();
773
if (!curr_otab->inUse)
774
iPoolTables.remove(curr_otab);
779
void MSOpenTablePool::returnToPool()
781
MSTableList::removeTablePool(this);
784
MSOpenTablePool *MSOpenTablePool::newPool(uint32_t db_id, uint32_t tab_id)
786
MSOpenTablePool *pool;
789
if (!(pool = new MSOpenTablePool())) {
790
CSException::throwOSError(CS_CONTEXT, ENOMEM);
793
pool->myPoolDB = MSDatabase::getDatabase(db_id);
794
pool->myPoolTableID = tab_id;
796
pool->myPoolTable = pool->myPoolDB->getTable(tab_id, false);
802
* ---------------------------------------------------------------
806
CSSyncOrderedList *MSTableList::gPoolListByID;
808
MSTableList::MSTableList()
812
MSTableList::~MSTableList()
816
void MSTableList::startUp()
818
new_(gPoolListByID, CSSyncOrderedList);
821
void MSTableList::shutDown()
824
gPoolListByID->clear();
825
gPoolListByID->release();
826
gPoolListByID = NULL;
830
class MSTableKey : public CSOrderKey {
832
uint32_t myKeyDatabaseID;
833
uint32_t myKeyTableID;
835
MSTableKey(): myKeyDatabaseID(0), myKeyTableID(0){ }
837
virtual ~MSTableKey() {
840
int compareKey(CSObject *key) {return CSObject::compareKey(key);}
841
virtual int compareKey(CSOrderKey *x) {
842
MSTableKey *key = (MSTableKey *) x;
845
if (myKeyDatabaseID < key->myKeyDatabaseID)
847
else if (myKeyDatabaseID > key->myKeyDatabaseID)
851
if (myKeyTableID < key->myKeyTableID)
853
else if (myKeyTableID > key->myKeyTableID)
860
static MSTableKey *newTableKey(uint32_t db_id, uint32_t tab_id)
864
if (!(key = new MSTableKey())) {
865
CSException::throwOSError(CS_CONTEXT, ENOMEM);
867
key->myKeyDatabaseID = db_id;
868
key->myKeyTableID = tab_id;
873
MSOpenTable *MSTableList::getOpenTableByID(uint32_t db_id, uint32_t tab_id)
875
MSOpenTablePool *pool;
876
MSOpenTable *otab = NULL;
880
lock_(gPoolListByID);
881
key.myKeyDatabaseID = db_id;
882
key.myKeyTableID = tab_id;
883
pool = (MSOpenTablePool *) gPoolListByID->find(&key);
886
pool = MSOpenTablePool::newPool(db_id, tab_id);
887
key_ptr = MSTableKey::newTableKey(db_id, tab_id);
888
gPoolListByID->add(key_ptr, pool);
890
if (!(otab = pool->getPoolTable())) {
891
otab = MSOpenTable::newOpenTable(pool);
892
pool->addOpenTable(otab);
895
unlock_(gPoolListByID);
899
MSOpenTable *MSTableList::getOpenTableForDB(uint32_t db_id)
901
return(MSTableList::getOpenTableByID(db_id, 0));
905
void MSTableList::releaseTable(MSOpenTable *otab)
907
MSOpenTablePool *pool;
910
lock_(gPoolListByID);
912
if ((pool = otab->myPool)) {
913
if (pool->isRemovingTP) {
914
pool->removeOpenTable(otab);
915
gPoolListByID->wakeup();
918
pool->returnOpenTable(otab);
921
unlock_(gPoolListByID);
925
bool MSTableList::removeTablePoolIfEmpty(MSOpenTablePool *pool)
928
if (pool->getSize() == 0) {
931
key.myKeyDatabaseID = pool->myPoolDB->myDatabaseID;
932
key.myKeyTableID = pool->myPoolTableID;
933
gPoolListByID->remove(&key);
934
/* TODO: Remove the table from the database, if it does not exist
942
void MSTableList::removeTablePool(MSOpenTablePool *pool)
945
lock_(gPoolListByID);
947
pool->isRemovingTP = true;
948
pool->removeOpenTablesNotInUse();
949
if (removeTablePoolIfEmpty(pool))
953
* Wait for the tables that are in use to be
956
gPoolListByID->wait();
958
unlock_(gPoolListByID);
963
* Close the pool associated with this open table.
965
void MSTableList::removeTablePool(MSOpenTable *otab)
967
MSOpenTablePool *pool;
970
key.myKeyDatabaseID = otab->getDB()->myDatabaseID;
971
key.myKeyTableID = otab->getTableID();
975
lock_(gPoolListByID);
977
if (!(pool = (MSOpenTablePool *) gPoolListByID->find(&key)))
979
pool->isRemovingTP = true;
980
pool->removeOpenTablesNotInUse();
981
if (removeTablePoolIfEmpty(pool))
984
* Wait for the tables that are in use to be
987
gPoolListByID->wait();
989
unlock_(gPoolListByID);
994
void MSTableList::removeDatabaseTables(MSDatabase *database)
996
MSOpenTablePool *pool;
1004
lock_(gPoolListByID);
1006
while ((pool = (MSOpenTablePool *) gPoolListByID->itemAt(idx))) {
1007
if (pool->myPoolDB == database) {
1012
unlock_(gPoolListByID);
1015
removeTablePool(pool);
1023
// lockTablePoolForDeletion() is only called to lock a pool for a table which is about to be removed.
1024
// When the pool is returned then it will be removed from the global pool list.
1025
MSOpenTablePool *MSTableList::lockTablePoolForDeletion(uint32_t db_id, uint32_t tab_id, CSString *db_name, CSString *tab_name)
1027
MSOpenTablePool *pool;
1036
key.myKeyDatabaseID = db_id;
1037
key.myKeyTableID = tab_id;
1039
lock_(gPoolListByID);
1042
if (!(pool = (MSOpenTablePool *) gPoolListByID->find(&key))) {
1043
char buffer[CS_EXC_MESSAGE_SIZE];
1045
cs_strcpy(CS_EXC_MESSAGE_SIZE, buffer, "Table is temporarily not available: ");
1046
cs_strcat(CS_EXC_MESSAGE_SIZE, buffer, db_name->getCString());
1048
cs_strcat(CS_EXC_MESSAGE_SIZE, buffer, ".");
1049
cs_strcat(CS_EXC_MESSAGE_SIZE, buffer, tab_name->getCString());
1051
CSException::throwException(CS_CONTEXT, MS_ERR_TABLE_LOCKED, buffer);
1053
pool->isRemovingTP = true;
1054
pool->removeOpenTablesNotInUse();
1055
if (pool->getSize() == 0) {
1056
// pool->retain(); Do not do this. The return to pool will free this by removing it from the list.
1060
* Wait for the tables that are in use to be
1063
gPoolListByID->wait();
1065
unlock_(gPoolListByID);
1074
MSOpenTablePool *MSTableList::lockTablePoolForDeletion(MSTable *tab)
1076
CSString *tab_name = NULL, *db_name;
1077
uint32_t db_id, tab_id;
1081
db_name = tab->myDatabase->myDatabaseName;
1084
tab_name = tab->myTableName;
1087
db_id = tab->myDatabase->myDatabaseID;
1088
tab_id = tab->myTableID;
1092
return_( lockTablePoolForDeletion(db_id, tab_id, db_name, tab_name));
1095
MSOpenTablePool *MSTableList::lockTablePoolForDeletion(MSOpenTable *otab)
1097
CSString *tab_name = NULL, *db_name;
1098
uint32_t db_id, tab_id;
1103
tab = otab->getDBTable();
1105
tab_name = tab->myTableName;
1109
db_name = otab->getDB()->myDatabaseName;
1112
db_id = otab->getDB()->myDatabaseID;
1113
tab_id = otab->getTableID();
1115
otab->returnToPool();
1117
return_( lockTablePoolForDeletion(db_id, tab_id, db_name, tab_name));
added
removed
plugin/pbms/src/open_table_ms.cc
