~pbms-core/pbms/5.11-beta

/* Copyright (c) 2008 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
 *
 * Original author: Paul McCullagh
 * Continued development: Barry Leslie
 *
 * 2007-07-10
 *
 * H&G2JCtL
 *
 * Network interface.
 *
 */

#include "CSConfig.h"
#include "CSGlobal.h"
#include "CSStrUtil.h"
#include "CSStorage.h"

#include "Defs_ms.h"
#include "Compactor_ms.h"
#include "OpenTable_ms.h"
#include "Repository_ms.h"

/*
 * ---------------------------------------------------------------
 * COMPACTOR THREAD
 */

MSCompactorThread::MSCompactorThread(time_t wait_time, MSDatabase *db):
CSDaemon(wait_time, NULL),
iCompactorDatabase(db)
{
}

MSCompactorThread::~MSCompactorThread()
{
	close();
}

void MSCompactorThread::close()
{
}

bool MSCompactorThread::doWork()
{
	bool				complete;
	MSRepository		*src_repo, *dst_repo;
	MSRepoFile			*src_file, *dst_file;
	csWord4				src_repo_id;
	MSBlobHeadRec		blob;
	off_t				src_offset;
	csWord2				head_size;
	csWord8				blob_size, blob_data_size;
	CSStringBuffer		*head;
	MSRepoPointersRec	ptr;
	u_int				table_ref_count;
	u_int				blob_ref_count;
	int					ref_count;
	size_t				ref_size;
	CSMutex				*lock;
	csWord4				tab_id;
	csWord8				blob_id;
	MSOpenTable			*otab;
	csWord4				repo_id;
	csWord8				repo_offset;
	csWord8				repo_blob_size;
	csWord2				repo_head_size;
	csWord2				tab_index;
	csWord1				status;

	enter_();
	retry:
	
#ifdef MS_COMPACTOR_POLLS
	if (!(src_repo = iCompactorDatabase->getRepoFullOfTrash(NULL)))
		return_(true);
#else
	myWaitTime = MS_DEFAULT_COMPACTOR_WAIT * 1000;  // Time in milli-seconds
	if (!(src_repo = iCompactorDatabase->getRepoFullOfTrash(&myWaitTime)))
		return_(true);
#endif
	frompool_(src_repo);
	src_file = src_repo->openRepoFile();
	push_(src_file);

	dst_repo = iCompactorDatabase->lockRepo(src_repo->myRepoFileSize - src_repo->myGarbageCount);
	frompool_(dst_repo);
	dst_file = dst_repo->openRepoFile();
	push_(dst_file);

	new_(head, CSStringBuffer(100));
	push_(head);

	complete = false;
	src_repo_id = src_repo->myRepoID;
	src_offset = src_repo->myRepoHeadSize;
	//printf("\nCompacting repo %d\n\n", src_repo_id);
	// For testing:
	{
		int blockit = 0;
		if (blockit) {
			release_(head);
			release_(dst_file);
			backtopool_(dst_repo);
			release_(src_file);
			backtopool_(src_repo);

			myWaitTime = 5 * 1000;  // Time in milli-seconds
			return_(true);
		}
	}
	while (src_offset < src_repo->myRepoFileSize) {			
		retry_loop:
		suspended();

		if (myMustQuit)
			goto quit;
		retry_read:
		
		// 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.
		lock = &src_repo->myRepoLock[src_offset % CS_REPO_REC_LOCK_COUNT];
		lock_(lock);
		if (src_file->read(&blob, src_offset, src_repo->myRepoBlobHeadSize, 0) < src_repo->myRepoBlobHeadSize) {
			unlock_(lock);
			break;
		}
		ref_size = CS_GET_DISK_1(blob.rb_ref_size_1);
		ref_count = CS_GET_DISK_2(blob.rb_ref_count_2);
		head_size = CS_GET_DISK_2(blob.rb_head_size_2);
		blob_size = CS_GET_DISK_6(blob.rb_blob_repo_size_6);
		blob_data_size = CS_GET_DISK_6(blob.rb_blob_data_size_6);
		status = CS_GET_DISK_1(blob.rb_status_1);
		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! */
			unlock_(lock);
			src_offset++;
			goto retry_read;
		}
		if (IN_USE_BLOB_STATUS(status)) {
			head->setLength(head_size);
			if (src_file->read(head->getBuffer(0), src_offset, head_size, 0) != head_size) {
				unlock_(lock);
				break;
			}

			table_ref_count = 0;
			blob_ref_count = 0;
			
			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:
						/* Check the reference: */
						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);

						otab = MSTableList::getOpenTableByID(iCompactorDatabase->myDatabaseID, tab_id);
						if (otab) {
							frompool_(otab);
							/* Ignore the return value (it will fail because auth_code is wrong!)!! */
							csWord4 auth_code = 0;
							otab->getDBTable()->readBlobHandle(otab, blob_id, &auth_code, &repo_id, &repo_offset, &repo_blob_size, &repo_head_size, false);
							backtopool_(otab);
							if (repo_id == src_repo_id &&
								repo_offset == src_offset &&
								repo_blob_size == blob_data_size &&
								repo_head_size == head_size)
								table_ref_count++;
							else
								/* Remove the reference: */
								CS_SET_DISK_2(ptr.rp_ref->rr_type_2, MS_BLOB_FREE_REF);
						}
						else
							CS_SET_DISK_2(ptr.rp_ref->rr_type_2, MS_BLOB_FREE_REF);
						break;
					case MS_BLOB_DELETE_REF:
						/* These are temporary references from the TempLog file. */
						/* We try to prevent this from happening, but it can! */
						csWord4			temp_log_id;
						csWord4			temp_log_offset;
						MSTempLogFile	*temp_log;

						temp_log_id = CS_GET_DISK_4(ptr.rp_temp_ref->tp_log_id_4);
						temp_log_offset = CS_GET_DISK_4(ptr.rp_temp_ref->tp_offset_4);
						if ((temp_log = iCompactorDatabase->openTempLogFile(temp_log_id, NULL, NULL))) {
							MSTempLogItemRec	log_item;
							csWord4				then;
							time_t				now;

							push_(temp_log);
							if (temp_log->read(&log_item, temp_log_offset, sizeof(MSTempLogItemRec), 0) == sizeof(MSTempLogItemRec)) {
								then = CS_GET_DISK_4(log_item.ti_time_4);
								now = time(NULL);
								if (now < then + MSTempLog::gTempBlobTimeout) {
									/* Wait for the BLOB to expire before we continue: */									
									release_(temp_log);
									unlock_(lock);

									/* Go to sleep until the problem has gone away! */
									lock_(this);
									suspendedWait(MSTempLog::adjustWaitTime(then, now));
									unlock_(this);
									goto retry_loop;
								}
							}
							release_(temp_log);
						}

						/* Remove the temp reference: */
						CS_SET_DISK_2(ptr.rp_ref->rr_type_2, MS_BLOB_FREE_REF);
						break;
					default:
						tab_index = CS_GET_DISK_2(ptr.rp_blob_ref->er_table_2);
						if (tab_index > ref_count || !tab_index) {
							/* Can't be true. Assume this is garbage! */
							unlock_(lock);
							src_offset++;
							goto retry_read;
						}
						blob_ref_count++;
						break;
				}
				ptr.rp_chars += ref_size;
			}

			if (table_ref_count && blob_ref_count) {
				/* Check the blob references again to make sure that they
				 * refer to valid table references.
				 */
				MSRepoTableRefPtr	tab_ref;

				blob_ref_count = 0;
				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_TABLE_REF:
						case MS_BLOB_DELETE_REF:
							break;
						default: // If it isn't one of the above we assume it is an blob ref. (er_table_2 can never have a value equal to one of the above REF type flags.)
								// It was already verified above that the index was with in range.
							tab_ref = (MSRepoTableRefPtr) (head->getBuffer(0) + src_repo->myRepoBlobHeadSize + (CS_GET_DISK_2(ptr.rp_blob_ref->er_table_2)-1) * ref_size);
							if (CS_GET_DISK_2(tab_ref->rr_type_2) == MS_BLOB_TABLE_REF)
								blob_ref_count++;
							break;
					}
					ptr.rp_chars += ref_size;
				}
			}

			if (blob_ref_count) {
				off_t dst_offset;

				dst_offset = dst_repo->myRepoFileSize;

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

				/* We have an engine reference, copy the BLOB over: */
				CSFile::transfer(dst_file, dst_offset + head_size, src_file, src_offset + head_size, blob_size, iCompactBuffer, MS_COMPACTOR_BUFFER_SIZE);

				/* If the BLOB has an alias update the alias index. */
				if (CS_GET_DISK_2(blob.rb_alias_offset_2)) {
					iCompactorDatabase->moveBlobAlias( src_repo_id, src_offset, CS_GET_DISK_4(blob.rb_alias_hash_4), dst_repo->myRepoID, dst_offset);
				}
				
				/* 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:
							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(iCompactorDatabase->myDatabaseID, tab_id))) {
								frompool_(otab);
								otab->getDBTable()->updateBlobHandle(otab, blob_id, dst_repo->myRepoID, dst_offset, 0);
								backtopool_(otab);
							}
							break;
						case MS_BLOB_DELETE_REF:
							break;
						default:
							break;
					}
					ptr.rp_chars += ref_size;
				}

				dst_repo->myRepoFileSize += head_size + blob_size;
			}
		}
		
		unlock_(lock);
		src_offset += head_size + blob_size;
	}

	src_repo->mustBeDeleted = true;
	complete = true;

	quit:
	release_(head);
	release_(dst_file);
	backtopool_(dst_repo);
	release_(src_file);
	backtopool_(src_repo);

	if (complete)
		iCompactorDatabase->removeRepo(src_repo_id, &myMustQuit);

	if (!myMustQuit)
		goto retry;
	return_(true);
}

void *MSCompactorThread::finalize()
{
	close();
	return NULL;
};