~drizzle-trunk/drizzle/development

/* Copyright (c) 2009 PrimeBase Technologies GmbH, Germany
 *
 * PrimeBase Media Stream for MySQL
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 *
 * Barry Leslie
 *
 * 2009-05-29
 *
 * H&G2JCtL
 *
 * Repository backup.
 *
 * The backup is done by creating a new database with the same name and ID in the 
 * backup location. Then the pbms_dump table in the source database is initialized
 * for a sequential scan for backup. This has the effect of locking all current repository
 * files. Then the equvalent of  'insert into dst_db.pbms_dump (select * from src_db.pbms_dump);'
 * is performed. 
 *
 */

#ifdef DRIZZLED
#include "config.h"
#include <drizzled/common.h>
#include <drizzled/session.h>
#include <drizzled/table.h>
#include <drizzled/message/table.pb.h>
#include "drizzled/charset_info.h"
#include <drizzled/table_proto.h>
#include <drizzled/field.h>
#include <drizzled/field/varstring.h>
#endif

#include "cslib/CSConfig.h"

#include <sys/types.h>
#include <inttypes.h>

#include "cslib/CSGlobal.h"
#include "cslib/CSStrUtil.h"
#include "cslib/CSStorage.h"

#include "defs_ms.h"
#include "system_table_ms.h"
#include "open_table_ms.h"
#include "table_ms.h"
#include "database_ms.h"
#include "repository_ms.h"
#include "backup_ms.h"
#include "transaction_ms.h"
#include "systab_variable_ms.h"
#include "systab_backup_ms.h"

uint32_t MSBackupInfo::gMaxInfoRef;
CSSyncSparseArray *MSBackupInfo::gBackupInfo;

//==========================================
MSBackupInfo::MSBackupInfo(	uint32_t id, 
							const char *name, 
							uint32_t db_id_arg, 
							time_t start, 
							time_t end, 
							bool _isDump, 
							const char *location, 
							uint32_t cloudRef_arg, 
							uint32_t cloudBackupNo_arg ):
	backupRefId(id),
	db_name(NULL),
	db_id(db_id_arg),
	startTime(start),
	completionTime(end),
	dump(_isDump),
	isRunning(false),
	backupLocation(NULL),
	cloudRef(cloudRef_arg),
	cloudBackupNo(cloudBackupNo_arg)
{
	db_name = CSString::newString(name);
	if (location && *location)		
		backupLocation = CSString::newString(location);		
}

//-------------------------------
MSBackupInfo::~MSBackupInfo()
{
	if (db_name)
		db_name->release();
	
	if (backupLocation)
		backupLocation->release();
}

//-------------------------------
void MSBackupInfo::startBackup(MSDatabase *pbms_db)
{
	MSDatabase *src_db;
	
	enter_();
	push_(pbms_db);
	
	src_db = MSDatabase::getDatabase(db_id);
	push_(src_db);
	
	startTime = time(NULL);
	
	src_db->startBackup(RETAIN(this));
	release_(src_db);
	
	isRunning = true;
	
	pop_(pbms_db);
	MSBackupTable::saveTable(pbms_db);
	exit_();
}

//-------------------------------
MSBackupInfo *MSBackupInfo::startDump(MSDatabase *db, uint32_t cloud_ref, uint32_t backup_no)
{
	MSBackupInfo *info;
	uint32_t ref_id;
	
	enter_();
	push_(db);
	lock_(gBackupInfo);
	
	ref_id = gMaxInfoRef++;
	new_(info, MSBackupInfo(ref_id, db->myDatabaseName->getCString(), db->myDatabaseID, time(NULL), 0, true, NULL, cloud_ref, backup_no));
	push_(info);
	
	gBackupInfo->set(ref_id, RETAIN(info));
	
	info->isRunning = true;

	pop_(info);
	unlock_(gBackupInfo);
	pop_(db);
	
	try_(a) {
		MSBackupTable::saveTable(db);
	}
	catch_(a);
	gBackupInfo->remove(ref_id);
	info->release();
	throw_();
	
	cont_(a);
	return_(info);
}
//-------------------------------
void MSBackupInfo::backupCompleted(MSDatabase *db)
{
	completionTime = time(NULL);	
	isRunning = false;
	MSBackupTable::saveTable(db);
}

//-------------------------------
void MSBackupInfo::backupTerminated(MSDatabase *db)
{
	enter_();
	push_(db);
	lock_(gBackupInfo);
	
	gBackupInfo->remove(backupRefId);
	unlock_(gBackupInfo);
	
	pop_(db);
	MSBackupTable::saveTable(db);
	exit_();
}

//==========================================
MSBackup::MSBackup():
CSDaemon(NULL),
bu_info(NULL),
bu_BackupList(NULL),
bu_Compactor(NULL),
bu_BackupRunning(false),
bu_State(BU_COMPLETED),
bu_SourceDatabase(NULL),
bu_Database(NULL),
bu_dst_dump(NULL),
bu_src_dump(NULL),
bu_size(0),
bu_completed(0),
bu_ID(0),
bu_start_time(0),
bu_TransactionManagerSuspended(false)
{
}

MSBackup *MSBackup::newMSBackup(MSBackupInfo *info)
{
	MSBackup *bu;
	enter_();
	
	push_(info);
	
	new_(bu, MSBackup());
	push_(bu);
	bu->bu_Database = MSDatabase::getBackupDatabase(RETAIN(info->backupLocation), RETAIN(info->db_name), info->db_id, true);
	pop_(bu);
	
	bu->bu_info = info;
	pop_(info);

	return_(bu);
}

void MSBackup::startBackup(MSDatabase *src_db)
{
	CSSyncVector	*repo_list;
	bool			compacting = false;
	MSRepository	*repo;
	enter_();

	try_(a) {
		bu_SourceDatabase = src_db;
		repo_list = bu_SourceDatabase->getRepositoryList();
		// Suspend the compactor before locking the list.
		bu_Compactor = bu_SourceDatabase->getCompactorThread();
		if (bu_Compactor) {
			bu_Compactor->retain();
			bu_Compactor->suspend();
		}

		// Build the list of repositories to be backed up.
		lock_(repo_list);

		new_(bu_BackupList, CSVector(repo_list->size()));
		for (uint32_t i = 0; i<repo_list->size(); i++) {
			if ((repo = (MSRepository *) repo_list->get(i))) {
				if (!repo->isRemovingFP && !repo->mustBeDeleted) {
					bu_BackupList->add(RETAIN(repo));
					if (repo->initBackup() == REPO_COMPACTING) 
						compacting = true; 
					
					if (!repo->myRepoHeadSize) {
						/* The file has not yet been opened, so the
						 * garbage count will not be known!
						 */
						MSRepoFile *repo_file;

						//repo->retain();
						//unlock_(myRepostoryList);
						//push_(repo);
						repo_file = repo->openRepoFile();
						repo_file->release();
						//release_(repo);
						//lock_(myRepostoryList);
						//goto retry;
					}
					
					bu_size += repo->myRepoFileSize; 

				}
			}
		}
		
		// Copy the table list to the backup database:
		uint32_t		next_tab = 0;
		MSTable		*tab;
		while ((tab = bu_SourceDatabase->getNextTable(&next_tab))) {
			push_(tab);
			bu_Database->addTable(tab->myTableID, tab->myTableName->getCString(), 0, false);
			release_(tab);
		}
		unlock_(repo_list);
		
		// Copy over any physical PBMS system tables.
		PBMSSystemTables::transferSystemTables(RETAIN(bu_Database), RETAIN(bu_SourceDatabase));

		// Load the system tables into the backup database. This will
		// initialize the database for cloud storage if required.
		PBMSSystemTables::loadSystemTables(RETAIN(bu_Database));
		
		// Set the cloud backup info.
		bu_Database->myBlobCloud->cl_setBackupInfo(RETAIN(bu_info));
		
		
		// Set the backup number in the pbms_variable tabe. (This is a hidden value.)
		// This value is used in case a drag and drop restore was done. When a data base is
		// first loaded this value is checked and if it is not zero then the backup record
		// will be read and any used to recover any BLOBs.
		// 
		char value[20];
		snprintf(value, 20, "%"PRIu32"", bu_info->getBackupRefId());
		MSVariableTable::setVariable(RETAIN(bu_Database), BACKUP_NUMBER_VAR, value);
		
		// Once the repositories are locked the compactor can be restarted
		// unless it is in the process of compacting a repository that is
		// being backed up.
		if (bu_Compactor && !compacting) {	
			bu_Compactor->resume();		
			bu_Compactor->release();		
			bu_Compactor = NULL;		
		}
		
		// Suspend the transaction writer while the backup is running.
		MSTransactionManager::suspend(true);
		bu_TransactionManagerSuspended = true;
		
		// Start the backup daemon thread.
		bu_ID = bu_start_time = time(NULL);
		start();
	}
	catch_(a) {
		completeBackup();
		throw_();
	}
	cont_(a);
	
	exit_();

}

void MSBackup::completeBackup()
{
	if (bu_TransactionManagerSuspended) {	
		MSTransactionManager::resume();
		bu_TransactionManagerSuspended = false;
	}

	if (bu_BackupList) {
		MSRepository *repo;		
		
		while (bu_BackupList->size()) {
			repo = (MSRepository *) bu_BackupList->take(0);
			if (repo) {				
				repo->backupCompleted();
				repo->release();				
			}
		}
		bu_BackupList->release();
		bu_BackupList = NULL;
	}
		
	if (bu_Compactor) {
		bu_Compactor->resume();
		bu_Compactor->release();
		bu_Compactor = NULL;
	}
	
	if (bu_Database) {
		if (bu_State == BU_COMPLETED)
			bu_Database->releaseBackupDatabase();
		else 
			MSDatabase::dropDatabase(bu_Database);
			
		bu_Database = NULL;
	}

	if (bu_SourceDatabase){
		if (bu_State == BU_COMPLETED) 
			bu_info->backupCompleted(bu_SourceDatabase);
		else 
			bu_info->backupTerminated(bu_SourceDatabase);
		
		bu_SourceDatabase = NULL;
		bu_info->release();
		bu_info = NULL;
	}
	
	bu_BackupRunning = false;
}

bool MSBackup::doWork()
{
	CSMutex				*my_lock;
	MSRepository		*src_repo, *dst_repo;
	MSRepoFile			*src_file, *dst_file;
	off64_t				src_offset, prev_offset;
	uint16_t				head_size;
	uint64_t				blob_size, blob_data_size;
	CSStringBuffer		*head;
	MSRepoPointersRec	ptr;
	uint32_t				table_ref_count;
	uint32_t				blob_ref_count;
	int					ref_count;
	size_t				ref_size;
	uint32_t				auth_code;
	uint32_t				tab_id;
	uint64_t				blob_id;
	MSOpenTable			*otab;
	uint32_t				src_repo_id;
	uint8_t				status;
	uint8_t				blob_storage_type;
	uint16_t				tab_index;
	uint32_t				mod_time;
	char				transferBuffer[MS_BACKUP_BUFFER_SIZE];
	CloudKeyRec			cloud_key;

	
	enter_();
	bu_BackupRunning = true;
	bu_State = BU_RUNNING; 

/*
	// For testing:
	{
		int blockit = 0;
		myWaitTime = 5 * 1000;  // Time in milli-seconds
		while (blockit)
			return_(true);
	}
*/
	
	try_(a) {
		new_(head, CSStringBuffer(100));
		push_(head);

		src_repo = (MSRepository*)bu_BackupList->get(0);
		while (src_repo && !myMustQuit) {
			src_offset = 0;
			src_file = src_repo->openRepoFile();
			push_(src_file);

			dst_repo = bu_Database->lockRepo(src_repo->myRepoFileSize - src_repo->myGarbageCount);
			frompool_(dst_repo);
			dst_file = dst_repo->openRepoFile();
			push_(dst_file);
			
			src_repo_id = src_repo->myRepoID;
			src_offset = src_repo->myRepoHeadSize;
			prev_offset = 0;
			while (src_offset < src_repo->myRepoFileSize) {	
	retry_read:
					
				bu_completed += src_offset - prev_offset;
				prev_offset = src_offset;
				suspended();

				if (myMustQuit)
					break;
				
				// A lock is required here because references and dereferences to the
				// BLOBs can result in the repository record being updated while 
				// it is being copied.
				my_lock = &src_repo->myRepoLock[src_offset % CS_REPO_REC_LOCK_COUNT];
				lock_(my_lock);
				head->setLength(src_repo->myRepoBlobHeadSize);
				if (src_file->read(head->getBuffer(0), src_offset, src_repo->myRepoBlobHeadSize, 0) < src_repo->myRepoBlobHeadSize) { 
					unlock_(my_lock);
					break;
				}
					
				ptr.rp_chars = head->getBuffer(0);
				ref_size = CS_GET_DISK_1(ptr.rp_head->rb_ref_size_1);
				ref_count = CS_GET_DISK_2(ptr.rp_head->rb_ref_count_2);
				head_size = CS_GET_DISK_2(ptr.rp_head->rb_head_size_2);
				blob_size = CS_GET_DISK_6(ptr.rp_head->rb_blob_repo_size_6);
				blob_data_size = CS_GET_DISK_6(ptr.rp_head->rb_blob_data_size_6);
				auth_code = CS_GET_DISK_4(ptr.rp_head->rb_auth_code_4);
				status = CS_GET_DISK_1(ptr.rp_head->rb_status_1);
				mod_time = CS_GET_DISK_4(ptr.rp_head->rb_mod_time_4);
				
				blob_storage_type = CS_GET_DISK_1(ptr.rp_head->rb_storage_type_1);
				if (blob_storage_type == MS_CLOUD_STORAGE) {
					MSRepoFile::getBlobKey(ptr.rp_head, &cloud_key);
				}

				// If the BLOB was modified after the start of the backup
				// then set the mod time to the backup time to ensure that
				// a backup for update will work correctly.
				if (mod_time > bu_start_time)
					CS_SET_DISK_4(ptr.rp_head->rb_mod_time_4, bu_start_time);
					
				// If the BLOB was moved during the time of this backup then copy
				// it to the backup location as a referenced BLOB.
				if ((status == MS_BLOB_MOVED)  && (bu_ID == (uint32_t) CS_GET_DISK_4(ptr.rp_head->rb_backup_id_4))) {
					status = MS_BLOB_REFERENCED;
					CS_SET_DISK_1(ptr.rp_head->rb_status_1, status);
				}
				
				// sanity check
				if ((blob_data_size == 0) || ref_count <= 0 || ref_size == 0 ||
					head_size < src_repo->myRepoBlobHeadSize + ref_count * ref_size ||
					!VALID_BLOB_STATUS(status)) {
					/* Can't be true. Assume this is garbage! */
					src_offset++;
					unlock_(my_lock);
					continue;
				}
				
				
				if ((status == MS_BLOB_REFERENCED) || (status == MS_BLOB_MOVED)) {
					head->setLength(head_size);
					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)) {
						unlock_(my_lock);
						break;
					}

					table_ref_count = 0;
					blob_ref_count = 0;
					
					// Loop through all the references removing temporary references 
					// and counting table and blob references.
					
					ptr.rp_chars = head->getBuffer(0) + src_repo->myRepoBlobHeadSize;
					for (int count = 0; count < ref_count; count++) {
						switch (CS_GET_DISK_2(ptr.rp_ref->rr_type_2)) {
							case MS_BLOB_FREE_REF:
								break;
							case MS_BLOB_TABLE_REF:
								// Unlike the compactor, table refs are not checked because
								// they do not yet exist in the backup database.
								table_ref_count++;
								break;
							case MS_BLOB_DELETE_REF:
								// These are temporary references from the TempLog file. 
								// They are not copied to the backup. 
								CS_SET_DISK_2(ptr.rp_ref->rr_type_2, MS_BLOB_FREE_REF);
								break;
							default:
								// Must be a BLOB reference
								
								tab_index = CS_GET_DISK_2(ptr.rp_blob_ref->er_table_2);
								if (tab_index && (tab_index <= ref_count)) {
									// Only committed references are backed up.
									if (IS_COMMITTED(CS_GET_DISK_8(ptr.rp_blob_ref->er_blob_ref_id_8))) {
										MSRepoTableRefPtr	tab_ref;
										tab_ref = (MSRepoTableRefPtr) (head->getBuffer(0) + src_repo->myRepoBlobHeadSize + (tab_index-1) * ref_size);
										if (CS_GET_DISK_2(tab_ref->rr_type_2) == MS_BLOB_TABLE_REF)
											blob_ref_count++;
									} else {
										CS_SET_DISK_2(ptr.rp_ref->rr_type_2, MS_BLOB_FREE_REF);
									}
								
								} else {
									/* Can't be true. Assume this is garbage! */
									src_offset++;
									unlock_(my_lock);
									goto retry_read;
								}
								break;
						}
						ptr.rp_chars += ref_size;
					}


					// If there are still blob references then the record needs to be backed up.
					if (table_ref_count && blob_ref_count) {

						off64_t dst_offset;

						dst_offset = dst_repo->myRepoFileSize;
						
						/* Write the header. */
						dst_file->write(head->getBuffer(0), dst_offset, head_size);

						/* Copy the BLOB over: */
						if (blob_storage_type == MS_CLOUD_STORAGE) { 
							bu_Database->myBlobCloud->cl_backupBLOB(&cloud_key);
						} else
							CSFile::transfer(dst_file, dst_offset + head_size, src_file, src_offset + head_size, blob_size, transferBuffer, MS_BACKUP_BUFFER_SIZE);
					
						/* Update the references: */
						ptr.rp_chars = head->getBuffer(0) + src_repo->myRepoBlobHeadSize;
						for (int count = 0; count < ref_count; count++) {
							switch (CS_GET_DISK_2(ptr.rp_ref->rr_type_2)) {
								case MS_BLOB_FREE_REF:
								case MS_BLOB_DELETE_REF:
									break;
								case MS_BLOB_TABLE_REF:
									tab_id = CS_GET_DISK_4(ptr.rp_tab_ref->tr_table_id_4);
									blob_id = CS_GET_DISK_6(ptr.rp_tab_ref->tr_blob_id_6);

									if ((otab = MSTableList::getOpenTableByID(bu_Database->myDatabaseID, tab_id))) {
										frompool_(otab);
										otab->getDBTable()->setBlobHandle(otab, blob_id, dst_repo->myRepoID, dst_offset, blob_size, head_size, auth_code);
//CSException::throwException(CS_CONTEXT, MS_ERR_NOT_IMPLEMENTED, "What if an error ocurred here!");

										backtopool_(otab);
									}
									break;
								default:
									break;
							}
							ptr.rp_chars += ref_size;
						}

						dst_repo->myRepoFileSize += head_size + blob_size;
					}
				}
				unlock_(my_lock);
				src_offset += head_size + blob_size;
			}
			bu_completed += src_offset - prev_offset;
			
			// close the destination repository and cleanup.
			release_(dst_file);
			backtopool_(dst_repo);
			release_(src_file);
			
			// release the source repository and get the next one in the list.
			src_repo->backupCompleted();
			bu_BackupList->remove(0);
			
			src_repo = (MSRepository*)bu_BackupList->get(0);
		}
				
		release_(head);
		if (myMustQuit)
			bu_State = BU_TERMINATED; 
		else
			bu_State = BU_COMPLETED; 
			
	}	
	
	catch_(a) {
		logException();
	}
	
	cont_(a);	
	completeBackup();
	myMustQuit = true;
	return_(true);
}

void *MSBackup::completeWork()
{
	if (bu_SourceDatabase || bu_BackupList || bu_Compactor || bu_info) {
		// We shouldn't be here
		CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "MSBackup::completeBackup() not called");
		if (bu_SourceDatabase) {
			 bu_SourceDatabase->release();
			 bu_SourceDatabase = NULL;
		}
			
		if (bu_BackupList) {
			 bu_BackupList->release();
			 bu_BackupList = NULL;
		}

			
		if (bu_Compactor) {
			 bu_Compactor->release();
			 bu_Compactor = NULL;
		}

			
		if (bu_info) {
			 bu_info->release();
			 bu_info = NULL;
		}

	}
	return NULL;
}