« back to all changes in this revision
Viewing changes to plugin/pbms/src/backup_ms.cc
- browse files at revision 1548.2.1
- compare with another revision
- compare with revision 2473
- download diff
- download tarball
- view history from revision 1548.2.1
- Committer: Barry.Leslie at PrimeBase
- Date: 2010-05-20 22:52:10 UTC
- mto: (1548.2.34 drizzle_pbms) (1643.1.1 build)
- mto: This revision was merged to the branch mainline in revision 1644.
- Revision ID: barry.leslie@primebase.com-20100520225210-b0q2jaovhwp2zluz
Added the PBMS daemon plugin.
(Augen zu und durch!)
(Augen zu und durch!)
- files added:
- plugin/pbms
- plugin/pbms/src
- plugin/pbms/src/cslib
added
removed
plugin/pbms/src/backup_ms.cc
1
/* Copyright (c) 2009 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
* Barry Leslie
20
*
21
* 2009-05-29
22
*
23
* H&G2JCtL
24
*
25
* Repository backup.
26
*
27
* The backup is done by creating a new database with the same name and ID in the
28
* backup location. Then the pbms_dump table in the source database is initialized
29
* for a sequential scan for backup. This has the effect of locking all current repository
30
* files. Then the equvalent of 'insert into dst_db.pbms_dump (select * from src_db.pbms_dump);'
31
* is performed.
32
*
33
*/
34
35
#ifdef DRIZZLED
36
#include <drizzled/server_includes.h>
37
#endif
38
39
#include <inttypes.h>
40
41
#include "CSConfig.h"
42
#include "CSGlobal.h"
43
#include "CSStrUtil.h"
44
#include "CSStorage.h"
45
46
#include "Defs_ms.h"
47
#include "SystemTable_ms.h"
48
#include "OpenTable_ms.h"
49
#include "Table_ms.h"
50
#include "Database_ms.h"
51
#include "Repository_ms.h"
52
#include "backup_ms.h"
53
#include "backup_ms.h"
54
#include "Transaction_ms.h"
55
#include "SysTab_variable.h"
56
#include "SysTab_backup.h"
57
58
uint32_t MSBackupInfo::gMaxInfoRef;
59
CSSyncSparseArray *MSBackupInfo::gBackupInfo;
60
61
//==========================================
62
MSBackupInfo::MSBackupInfo( uint32_t id,
63
const char *name,
64
uint32_t db_id_arg,
65
time_t start,
66
time_t end,
67
bool isDump,
68
const char *location,
69
uint32_t cloudRef_arg,
70
uint32_t cloudBackupNo_arg ):
71
backupRefId(id),
72
db_name(NULL),
73
db_id(db_id_arg),
74
startTime(start),
75
completionTime(end),
76
isRunning(false),
77
dump(isDump),
78
backupLocation(NULL),
79
cloudRef(cloudRef_arg),
80
cloudBackupNo(cloudBackupNo_arg)
81
{
82
db_name = CSString::newString(name);
83
if (location && *location)
84
backupLocation = CSString::newString(location);
85
}
86
87
//-------------------------------
88
MSBackupInfo::~MSBackupInfo()
89
{
90
if (db_name)
91
db_name->release();
92
93
if (backupLocation)
94
backupLocation->release();
95
}
96
97
//-------------------------------
98
void MSBackupInfo::startBackup(MSDatabase *pbms_db)
99
{
100
MSDatabase *src_db;
101
102
enter_();
103
push_(pbms_db);
104
105
src_db = MSDatabase::getDatabase(db_id);
106
push_(src_db);
107
108
startTime = time(NULL);
109
110
src_db->startBackup(RETAIN(this));
111
release_(src_db);
112
113
isRunning = true;
114
115
pop_(pbms_db);
116
MSBackupTable::saveTable(pbms_db);
117
exit_();
118
}
119
120
//-------------------------------
121
MSBackupInfo *MSBackupInfo::startDump(MSDatabase *db, uint32_t cloud_ref, uint32_t backup_no)
122
{
123
MSBackupInfo *info;
124
uint32_t ref_id;
125
126
enter_();
127
push_(db);
128
lock_(gBackupInfo);
129
130
ref_id = gMaxInfoRef++;
131
new_(info, MSBackupInfo(ref_id, db->myDatabaseName->getCString(), db->myDatabaseID, time(NULL), 0, true, NULL, cloud_ref, backup_no));
132
push_(info);
133
134
gBackupInfo->set(ref_id, RETAIN(info));
135
136
info->isRunning = true;
137
138
pop_(info);
139
unlock_(gBackupInfo);
140
pop_(db);
141
142
try_(a) {
143
MSBackupTable::saveTable(db);
144
}
145
catch_(a);
146
gBackupInfo->remove(ref_id);
147
info->release();
148
throw_();
149
150
cont_(a);
151
return_(info);
152
}
153
//-------------------------------
154
void MSBackupInfo::backupCompleted(MSDatabase *db)
155
{
156
completionTime = time(NULL);
157
isRunning = false;
158
MSBackupTable::saveTable(db);
159
}
160
161
//-------------------------------
162
void MSBackupInfo::backupTerminated(MSDatabase *db)
163
{
164
enter_();
165
push_(db);
166
lock_(gBackupInfo);
167
168
gBackupInfo->remove(backupRefId);
169
unlock_(gBackupInfo);
170
171
pop_(db);
172
MSBackupTable::saveTable(db);
173
exit_();
174
}
175
176
//==========================================
177
MSBackup::MSBackup():
178
CSDaemon(NULL),
179
bu_SourceDatabase(NULL),
180
bu_Database(NULL),
181
bu_info(NULL),
182
bu_BackupList(NULL),
183
bu_Compactor(NULL),
184
bu_completed(0),
185
bu_BackupRunning(false),
186
bu_size(0)
187
{
188
}
189
190
MSBackup::~MSBackup()
191
{
192
193
if (bu_SourceDatabase || bu_BackupList || bu_Compactor || bu_info) {
194
// We shouldn't be here
195
CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "MSBackup::completeBackup() not called");
196
if (bu_SourceDatabase)
197
bu_SourceDatabase->release();
198
199
if (bu_BackupList)
200
bu_BackupList->release();
201
202
if (bu_Compactor)
203
bu_Compactor->release();
204
205
if (bu_info)
206
bu_info->release();
207
}
208
209
}
210
211
MSBackup *MSBackup::newMSBackup(MSBackupInfo *info)
212
{
213
MSBackup *bu;
214
enter_();
215
216
push_(info);
217
218
new_(bu, MSBackup());
219
push_(bu);
220
bu->bu_Database = MSDatabase::getBackupDatabase(RETAIN(info->backupLocation), RETAIN(info->db_name), info->db_id, true);
221
pop_(bu);
222
223
bu->bu_info = info;
224
pop_(info);
225
226
return_(bu);
227
}
228
229
void MSBackup::startBackup(MSDatabase *src_db)
230
{
231
CSSyncVector *repo_list;
232
bool compacting = false;
233
MSRepository *repo;
234
enter_();
235
236
try_(a) {
237
bu_SourceDatabase = src_db;
238
repo_list = bu_SourceDatabase->getRepositoryList();
239
// Suspend the compactor before locking the list.
240
bu_Compactor = bu_SourceDatabase->getCompactorThread();
241
if (bu_Compactor) {
242
bu_Compactor->retain();
243
bu_Compactor->suspend();
244
}
245
246
// Build the list of repositories to be backed up.
247
lock_(repo_list);
248
249
new_(bu_BackupList, CSVector(repo_list->size()));
250
for (u_int i = 0; i<repo_list->size(); i++) {
251
if ((repo = (MSRepository *) repo_list->get(i))) {
252
if (!repo->isRemovingFP && !repo->mustBeDeleted) {
253
bu_BackupList->add(RETAIN(repo));
254
if (repo->initBackup() == REPO_COMPACTING)
255
compacting = true;
256
257
if (!repo->myRepoHeadSize) {
258
/* The file has not yet been opened, so the
259
* garbage count will not be known!
260
*/
261
MSRepoFile *repo_file;
262
263
//repo->retain();
264
//unlock_(myRepostoryList);
265
//push_(repo);
266
repo_file = repo->openRepoFile();
267
repo_file->release();
268
//release_(repo);
269
//lock_(myRepostoryList);
270
//goto retry;
271
}
272
273
bu_size += repo->myRepoFileSize;
274
275
}
276
}
277
}
278
279
// Copy the table list to the backup database:
280
uint32_t next_tab = 0;
281
MSTable *tab;
282
while ((tab = bu_SourceDatabase->getNextTable(&next_tab))) {
283
push_(tab);
284
bu_Database->addTable(tab->myTableID, tab->myTableName->getCString(), 0, false);
285
release_(tab);
286
}
287
unlock_(repo_list);
288
289
// Copy over any physical PBMS system tables.
290
pbms_transfer_ststem_tables(RETAIN(bu_Database), RETAIN(bu_SourceDatabase));
291
292
// Load the system tables into the backup database. This will
293
// initialize the database for cloud storage if required.
294
pbms_load_system_tables(RETAIN(bu_Database));
295
296
// Set the cloud backup info.
297
bu_Database->myBlobCloud->cl_setBackupInfo(RETAIN(bu_info));
298
299
300
// Set the backup number in the pbms_variable tabe. (This is a hidden value.)
301
// This value is used in case a drag and drop restore was done. When a data base is
302
// first loaded this value is checked and if it is not zero then the backup record
303
// will be read and any used to recover any BLOBs.
304
//
305
char value[20];
306
snprintf(value, 20, "%"PRIu32"", bu_info->getBackupRefId());
307
MSVariableTable::setVariable(RETAIN(bu_Database), BACKUP_NUMBER_VAR, value);
308
309
// Once the repositories are locked the compactor can be restarted
310
// unless it is in the process of compacting a repository that is
311
// being backed up.
312
if (bu_Compactor && !compacting) {
313
bu_Compactor->resume();
314
bu_Compactor->release();
315
bu_Compactor = NULL;
316
}
317
318
// Suspend the transaction writer while the backup is running.
319
MSTransactionManager::suspend(true);
320
bu_TransactionManagerSuspended = true;
321
322
// Start the backup daemon thread.
323
bu_ID = bu_start_time = time(NULL);
324
start();
325
}
326
catch_(a) {
327
completeBackup();
328
throw_();
329
}
330
cont_(a);
331
332
exit_();
333
334
}
335
336
void MSBackup::completeBackup()
337
{
338
if (bu_TransactionManagerSuspended) {
339
MSTransactionManager::resume();
340
bu_TransactionManagerSuspended = false;
341
}
342
343
if (bu_BackupList) {
344
MSRepository *repo;
345
346
while (bu_BackupList->size()) {
347
repo = (MSRepository *) bu_BackupList->take(0);
348
if (repo) {
349
repo->backupCompleted();
350
repo->release();
351
}
352
}
353
bu_BackupList->release();
354
bu_BackupList = NULL;
355
}
356
357
if (bu_Compactor) {
358
bu_Compactor->resume();
359
bu_Compactor->release();
360
bu_Compactor = NULL;
361
}
362
363
if (bu_Database) {
364
if (bu_State == BU_COMPLETED)
365
bu_Database->releaseBackupDatabase();
366
else
367
MSDatabase::dropDatabase(bu_Database);
368
369
bu_Database = NULL;
370
}
371
372
if (bu_SourceDatabase){
373
if (bu_State == BU_COMPLETED)
374
bu_info->backupCompleted(bu_SourceDatabase);
375
else
376
bu_info->backupTerminated(bu_SourceDatabase);
377
378
bu_SourceDatabase = NULL;
379
bu_info->release();
380
bu_info = NULL;
381
}
382
383
bu_BackupRunning = false;
384
}
385
386
bool MSBackup::doWork()
387
{
388
CSMutex *lock;
389
MSRepository *src_repo, *dst_repo;
390
MSRepoFile *src_file, *dst_file;
391
off_t src_offset, prev_offset;
392
uint16_t head_size;
393
uint64_t blob_size, blob_data_size;
394
CSStringBuffer *head;
395
MSRepoPointersRec ptr;
396
u_int table_ref_count;
397
u_int blob_ref_count;
398
int ref_count;
399
size_t ref_size;
400
uint32_t auth_code;
401
uint32_t tab_id;
402
uint64_t blob_id;
403
MSOpenTable *otab;
404
uint32_t src_repo_id;
405
uint8_t status;
406
uint8_t blob_storage_type;
407
uint16_t tab_index;
408
uint32_t mod_time;
409
char transferBuffer[MS_BACKUP_BUFFER_SIZE];
410
CloudKeyRec cloud_key;
411
412
413
enter_();
414
bu_BackupRunning = true;
415
bu_State = BU_RUNNING;
416
417
/*
418
// For testing:
419
{
420
int blockit = 0;
421
myWaitTime = 5 * 1000; // Time in milli-seconds
422
while (blockit)
423
return_(true);
424
}
425
*/
426
427
try_(a) {
428
new_(head, CSStringBuffer(100));
429
push_(head);
430
431
src_repo = (MSRepository*)bu_BackupList->get(0);
432
while (src_repo && !myMustQuit) {
433
src_offset = 0;
434
src_file = src_repo->openRepoFile();
435
push_(src_file);
436
437
dst_repo = bu_Database->lockRepo(src_repo->myRepoFileSize - src_repo->myGarbageCount);
438
frompool_(dst_repo);
439
dst_file = dst_repo->openRepoFile();
440
push_(dst_file);
441
442
src_repo_id = src_repo->myRepoID;
443
src_offset = src_repo->myRepoHeadSize;
444
prev_offset = 0;
445
while (src_offset < src_repo->myRepoFileSize) {
446
retry_read:
447
448
bu_completed += src_offset - prev_offset;
449
prev_offset = src_offset;
450
suspended();
451
452
if (myMustQuit)
453
break;
454
455
// A lock is required here because references and dereferences to the
456
// BLOBs can result in the repository record being updated while
457
// it is being copied.
458
lock = &src_repo->myRepoLock[src_offset % CS_REPO_REC_LOCK_COUNT];
459
lock_(lock);
460
head->setLength(src_repo->myRepoBlobHeadSize);
461
if (src_file->read(head->getBuffer(0), src_offset, src_repo->myRepoBlobHeadSize, 0) < src_repo->myRepoBlobHeadSize) {
462
unlock_(lock);
463
break;
464
}
465
466
ptr.rp_chars = head->getBuffer(0);
467
ref_size = CS_GET_DISK_1(ptr.rp_head->rb_ref_size_1);
468
ref_count = CS_GET_DISK_2(ptr.rp_head->rb_ref_count_2);
469
head_size = CS_GET_DISK_2(ptr.rp_head->rb_head_size_2);
470
blob_size = CS_GET_DISK_6(ptr.rp_head->rb_blob_repo_size_6);
471
blob_data_size = CS_GET_DISK_6(ptr.rp_head->rb_blob_data_size_6);
472
auth_code = CS_GET_DISK_4(ptr.rp_head->rb_auth_code_4);
473
status = CS_GET_DISK_1(ptr.rp_head->rb_status_1);
474
mod_time = CS_GET_DISK_4(ptr.rp_head->rb_mod_time_4);
475
476
blob_storage_type = CS_GET_DISK_1(ptr.rp_head->rb_storage_type_1);
477
if (blob_storage_type == MS_CLOUD_STORAGE) {
478
MSRepoFile::getBlobKey(ptr.rp_head, &cloud_key);
479
}
480
481
// If the BLOB was modified after the start of the backup
482
// then set the mod time to the backup time to ensure that
483
// a backup for update will work correctly.
484
if (mod_time > bu_start_time)
485
CS_SET_DISK_4(ptr.rp_head->rb_mod_time_4, bu_start_time);
486
487
// If the BLOB was moved during the time of this backup then copy
488
// it to the backup location as a referenced BLOB.
489
if ((status == MS_BLOB_MOVED) && (bu_ID == (uint32_t) CS_GET_DISK_4(ptr.rp_head->rb_backup_id_4))) {
490
status = MS_BLOB_REFERENCED;
491
CS_SET_DISK_1(ptr.rp_head->rb_status_1, status);
492
}
493
494
// sanity check
495
if ((blob_data_size == 0) || ref_count <= 0 || ref_size == 0 ||
496
head_size < src_repo->myRepoBlobHeadSize + ref_count * ref_size ||
497
!VALID_BLOB_STATUS(status)) {
498
/* Can't be true. Assume this is garbage! */
499
src_offset++;
500
unlock_(lock);
501
continue;
502
}
503
504
505
if ((status == MS_BLOB_REFERENCED) || (status == MS_BLOB_MOVED)) {
506
head->setLength(head_size);
507
if (src_file->read(head->getBuffer(0) + src_repo->myRepoBlobHeadSize, src_offset + src_repo->myRepoBlobHeadSize, head_size - src_repo->myRepoBlobHeadSize, 0) != (head_size- src_repo->myRepoBlobHeadSize)) {
508
unlock_(lock);
509
break;
510
}
511
512
table_ref_count = 0;
513
blob_ref_count = 0;
514
515
// Loop through all the references removing temporary references
516
// and counting table and blob references.
517
518
ptr.rp_chars = head->getBuffer(0) + src_repo->myRepoBlobHeadSize;
519
for (int count = 0; count < ref_count; count++) {
520
switch (CS_GET_DISK_2(ptr.rp_ref->rr_type_2)) {
521
case MS_BLOB_FREE_REF:
522
break;
523
case MS_BLOB_TABLE_REF:
524
// Unlike the compactor, table refs are not checked because
525
// they do not yet exist in the backup database.
526
table_ref_count++;
527
break;
528
case MS_BLOB_DELETE_REF:
529
// These are temporary references from the TempLog file.
530
// They are not copied to the backup.
531
CS_SET_DISK_2(ptr.rp_ref->rr_type_2, MS_BLOB_FREE_REF);
532
break;
533
default:
534
// Must be a BLOB reference
535
536
tab_index = CS_GET_DISK_2(ptr.rp_blob_ref->er_table_2);
537
if (tab_index && (tab_index <= ref_count)) {
538
// Only committed references are backed up.
539
if (IS_COMMITTED(CS_GET_DISK_8(ptr.rp_blob_ref->er_blob_ref_id_8))) {
540
MSRepoTableRefPtr tab_ref;
541
tab_ref = (MSRepoTableRefPtr) (head->getBuffer(0) + src_repo->myRepoBlobHeadSize + (tab_index-1) * ref_size);
542
if (CS_GET_DISK_2(tab_ref->rr_type_2) == MS_BLOB_TABLE_REF)
543
blob_ref_count++;
544
} else {
545
CS_SET_DISK_2(ptr.rp_ref->rr_type_2, MS_BLOB_FREE_REF);
546
}
547
548
} else {
549
/* Can't be true. Assume this is garbage! */
550
src_offset++;
551
unlock_(lock);
552
goto retry_read;
553
}
554
break;
555
}
556
ptr.rp_chars += ref_size;
557
}
558
559
560
// If there are still blob references then the record needs to be backed up.
561
if (table_ref_count && blob_ref_count) {
562
563
off_t dst_offset;
564
565
dst_offset = dst_repo->myRepoFileSize;
566
567
/* Write the header. */
568
dst_file->write(head->getBuffer(0), dst_offset, head_size);
569
570
/* Copy the BLOB over: */
571
if (blob_storage_type == MS_CLOUD_STORAGE) {
572
bu_Database->myBlobCloud->cl_backupBLOB(&cloud_key);
573
} else
574
CSFile::transfer(dst_file, dst_offset + head_size, src_file, src_offset + head_size, blob_size, transferBuffer, MS_BACKUP_BUFFER_SIZE);
575
576
/* Update the references: */
577
ptr.rp_chars = head->getBuffer(0) + src_repo->myRepoBlobHeadSize;
578
for (int count = 0; count < ref_count; count++) {
579
switch (CS_GET_DISK_2(ptr.rp_ref->rr_type_2)) {
580
case MS_BLOB_FREE_REF:
581
case MS_BLOB_DELETE_REF:
582
break;
583
case MS_BLOB_TABLE_REF:
584
tab_id = CS_GET_DISK_4(ptr.rp_tab_ref->tr_table_id_4);
585
blob_id = CS_GET_DISK_6(ptr.rp_tab_ref->tr_blob_id_6);
586
587
if ((otab = MSTableList::getOpenTableByID(bu_Database->myDatabaseID, tab_id))) {
588
frompool_(otab);
589
otab->getDBTable()->setBlobHandle(otab, blob_id, dst_repo->myRepoID, dst_offset, blob_size, head_size, auth_code);
590
//CSException::throwException(CS_CONTEXT, MS_ERR_NOT_IMPLEMENTED, "What if an error ocurred here!");
591
592
backtopool_(otab);
593
}
594
break;
595
default:
596
break;
597
}
598
ptr.rp_chars += ref_size;
599
}
600
601
dst_repo->myRepoFileSize += head_size + blob_size;
602
}
603
}
604
unlock_(lock);
605
src_offset += head_size + blob_size;
606
}
607
bu_completed += src_offset - prev_offset;
608
609
// close the destination repository and cleanup.
610
release_(dst_file);
611
backtopool_(dst_repo);
612
release_(src_file);
613
614
// release the source repository and get the next one in the list.
615
src_repo->backupCompleted();
616
bu_BackupList->remove(0);
617
618
src_repo = (MSRepository*)bu_BackupList->get(0);
619
}
620
621
release_(head);
622
if (myMustQuit)
623
bu_State = BU_TERMINATED;
624
else
625
bu_State = BU_COMPLETED;
626
627
}
628
629
catch_(a) {
630
logException();
631
}
632
633
cont_(a);
634
completeBackup();
635
myMustQuit = true;
636
return_(true);
637
}
638
639
void *MSBackup::finalize()
640
{
641
return NULL;
642
};
Loggerhead is a web-based interface for Breezy
Version: 2.0.1