~drizzle-trunk/drizzle/development

/* 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-05-20

 *

 * H&G2JCtL

 *

 * Table handler.

 *

 */

#ifdef USE_PRAGMA_IMPLEMENTATION

#pragma implementation				// gcc: Class implementation

#endif

#ifdef DRIZZLED

#include <drizzled/server_includes.h>

#include <drizzled/handler.h>

#include <drizzled/table.h>

#include <drizzled/current_session.h>

#include <drizzled/plugin/storage_engine.h>

#include <drizzled/plugin.h>

#define my_strdup(a,b) strdup(a)

/*

#include <sys/stat.h>

#include <drizzled/common_server.h>

#include <drizzled/data_home.h>

#include <drizzled/error.h>

#include <drizzled/handler.h>

#include <drizzled/plugin/storage_engine.h>

*/

#include "CSConfig.h"

#else

#include "CSConfig.h"

#include "mysql_priv.h"

#include <mysql/plugin.h>

#include <my_dir.h>

#endif 

#include <stdlib.h>

#include <time.h>

#include "Defs_ms.h"

#include "CSDefs.h"

#include "CSObject.h"

#include "CSGlobal.h"

#include "CSThread.h"

#include "CSStrUtil.h"

#include "CSTest.h"

#include "CSLog.h"

#include "Engine_ms.h"	

#include "ha_pbms.h"

#include "Network_ms.h"

#include "ConnectionHandler_ms.h"

#include "OpenTable_ms.h"

#include "Database_ms.h"

#include "TempLog_ms.h"

#include "Util_ms.h"

#include "SystemTable_ms.h"

#include "ms_mysql.h"

#include "Discover_ms.h"

#include "metadata_ms.h"

#include "Transaction_ms.h"

#include "SysTab_httpheader.h"

/* Note: 'new' used here is NOT CSObject::new which is a DEBUG define*/

#ifdef new

#undef new

#endif

#ifndef PBMS_PORT

#define PBMS_PORT 8080

#endif

static int		pbms_port = PBMS_PORT; 

static char		*pbms_repository_threshold;

static char		*pbms_temp_log_threshold;

static char		*pbms_http_metadata_headers;

#ifdef DRIZZLED

class PBMSTableNameIterator: public TableNameIteratorImplementation

{

private:

  uint32_t current_name;

public:

  PBMSTableNameIterator(const std::string &database)

    : TableNameIteratorImplementation(database), current_name(0)

    {};

   int next(std::string *name);

};

int PBMSTableNameIterator::next(string *name)

{

	const char *tab_name = pbms_getSysTableName(current_name++);

	if (!tab_name) 

		return -1;

    if (name)

      name->assign(tab_name);

	return 0;

}

int pbms_discover_system_tables(const char *name, drizzled::message::Table *table);

class PBMSStorageEngine : public StorageEngine {

public:

	PBMSStorageEngine(std::string name_arg)

	: StorageEngine(name_arg, HTON_NO_FLAGS | HTON_HIDDEN, 4 /*savepoint_offset*/, false /*support_2pc*/) {}

	int close_connection(Session *);

	int commit(Session *, bool);

	int rollback(Session *, bool);

	handler *create(TABLE_SHARE *, MEM_ROOT *);

	void drop_database(char *);

	int savepoint_set(Session *thd, void *sv);

	int savepoint_rollback(Session *thd, void *sv);

	int savepoint_release(Session *thd, void *sv);

	const char **bas_ext() const;

	int createTableImplementation(Session*, const char *path, Table *,HA_CREATE_INFO *, drizzled::message::Table*)

	{

		if (pbms_is_Systable(cs_last_name_of_path(path)))

			return(0);

		/* Create only works for system tables. */

		return( HA_ERR_WRONG_COMMAND );

	}

	int deleteTableImplementation(Session*, const string table_name) { return 0;}

	int getTableProtoImplementation(const char* path, drizzled::message::Table *table_proto)

	{

		int err = pbms_discover_system_tables(path, table_proto);

		if (err)

			return err;

		return EEXIST;

	}

	TableNameIteratorImplementation* tableNameIterator(const std::string &database)

	{

		return new PBMSTableNameIterator(database);

	}

	int renameTableImplementation(Session*,

											  const char *from, const char *to)

	{

	  return -1;

	}

};

PBMSStorageEngine	*pbms_hton;

#else

handlerton		*pbms_hton;

#endif

static const char *ha_pbms_exts[] = {

	NullS

};

/*

 * ---------------------------------------------------------------

 * UTILITIES

 */

void pbms_take_part_in_transaction(void *thread)

{

	THD			*thd;

	if ((thd = (THD *) thread)) {

		trans_register_ha(thd, true, pbms_hton); 

	}

}

#ifdef DRIZZLED

const char **PBMSStorageEngine::bas_ext() const

#else

const char **ha_pbms::bas_ext() const

#endif

{

	return ha_pbms_exts;

}

#ifdef DRIZZLED

int PBMSStorageEngine::close_connection(Session *thd)

{

	PBMSStorageEngine * const hton = this;

#else

static int pbms_close_connection(handlerton *hton, THD* thd)

{

#endif

	CSThread	*self;

	self = CSThread::getSelf();

	if (self && self->pbms_api_owner)

		return 0;

	if (thd) {

		if ((self = (CSThread *) *thd_ha_data(thd, pbms_hton))) {

			*thd_ha_data(thd, pbms_hton) = NULL;

			CSThread::setSelf(self);

			CSThread::detach(self);

		}

	}

	else {

		self = CSThread::getSelf();

		CSThread::detach(self);

	}

	return 0;

}

static int pbms_enter_conn(void *thread, CSThread **r_self, PBMSResultPtr result)

{

	THD			*thd;

	CSThread	*self;

	self = CSThread::getSelf();

	if (!self) {	

		if ((thd = (THD *) thread)) {

			if (!(self = (CSThread *) *thd_ha_data(((THD *) thd), pbms_hton))) {

				if (!(self = CSThread::newCSThread()))

					return pbms_os_error_result(CS_CONTEXT, ENOMEM, result);

				if (!CSThread::attach(self))

					return pbms_exception_to_result(&self->myException, result);

				*thd_ha_data(thd, pbms_hton) = self;

			}

			else {

				if (!CSThread::setSelf(self))

					return pbms_exception_to_result(&self->myException, result);

			}

		}

		else {

			if (!(self = CSThread::newCSThread()))

				return pbms_os_error_result(CS_CONTEXT, ENOMEM, result);

			if (!CSThread::attach(self))

				return pbms_exception_to_result(&self->myException, result);

		}

	}

	*r_self = self;

	return MS_OK;

}

int pbms_enter_conn_no_thd(CSThread **r_self, PBMSResultPtr result)

{

	return pbms_enter_conn(current_thd, r_self, result);

}

void pbms_exit_conn()

{

	THD			*thd = (THD *) current_thd;

	CSThread	*self;

	self = CSThread::getSelf();

	if (self && self->pbms_api_owner)

		return;

	if (thd)

		CSThread::setSelf(NULL);

	else {

		self = CSThread::getSelf();

		CSThread::detach(self);

	}

}

int pbms_exception_to_result(CSException *e, PBMSResultPtr result)

{

	const char *context, *trace;

	result->mr_code = e->getErrorCode();

	cs_strcpy(MS_RESULT_MESSAGE_SIZE, result->mr_message, e->getMessage());

	context = e->getContext();

	trace = e->getStackTrace();

	if (context && *context) {

		cs_strcpy(MS_RESULT_STACK_SIZE, result->mr_stack, context);

		if (trace && *trace)

			cs_strcat(MS_RESULT_STACK_SIZE, result->mr_stack, "\n");

	}

	else

		*result->mr_stack = 0;

	if (trace && *trace)

		cs_strcat(MS_RESULT_STACK_SIZE, result->mr_stack, trace);

	return MS_ERR_ENGINE;

}

int pbms_os_error_result(const char *func, const char *file, int line, int err, PBMSResultPtr result)

{

	CSException e;

	e.initOSError(func, file, line, err);

	return pbms_exception_to_result(&e, result);

}

int pbms_error_result(const char *func, const char *file, int line, int err, const char *message, PBMSResultPtr result)

{

	CSException e;

	e.initException(func, file, line, err, message);

	return pbms_exception_to_result(&e, result);

}

/*

 * ---------------------------------------------------------------

 * HANDLER INTERFACE

 */

#ifdef DRIZZLED

handler *PBMSStorageEngine::create(TABLE_SHARE *table, MEM_ROOT *mem_root)

{

	PBMSStorageEngine * const hton = this;

#else

static handler *pbms_create_handler(handlerton *hton, TABLE_SHARE *table, MEM_ROOT *mem_root)

{

#endif

	return new (mem_root) ha_pbms(hton, table);

}

#ifdef DRIZZLED

int PBMSStorageEngine::commit(Session *thd, bool all)

{

	PBMSStorageEngine * const hton = this;

#else

static int pbms_commit(handlerton *hton, THD *thd, bool all)

{

#endif

	int			err = 0;

	CSThread	*self;

	PBMSResultRec result;

	if (pbms_enter_conn_no_thd(&self, &result))

		return 0;

	inner_();

	try_(a) {

		MSTransactionManager::commit();

	}

	catch_(a) {

		err = pbms_exception_to_result(&self->myException, &result);

	}

	cont_(a);

	return_(err);

}

#ifdef DRIZZLED

int PBMSStorageEngine::rollback(Session *thd, bool all)

{

	PBMSStorageEngine * const hton = this;

#else

static int pbms_rollback(handlerton *hton, THD *thd, bool all)

{

#endif

	int			err = 0;

	CSThread	*self;

	PBMSResultRec result;

	if (pbms_enter_conn_no_thd(&self, &result))

		return 0;

	inner_();

	try_(a) {

		MSTransactionManager::rollback();

	}

	catch_(a) {

		err = pbms_exception_to_result(&self->myException, &result);

	}

	cont_(a);

	return_(err);

}

#ifdef DRIZZLED

int PBMSStorageEngine::savepoint_set(Session *thd, void *sv)

{

	PBMSStorageEngine * const hton = this;

#else

static int pbms_savepoint_set(handlerton *hton, THD *thd, void *sv)

{

#endif

	int			err = 0;

	CSThread	*self;

	PBMSResultRec result;

	if (pbms_enter_conn_no_thd(&self, &result))

		return 0;

	*((uint32_t*)sv) = self->myStmtCount;

	return 0;	

}

#ifdef DRIZZLED

int PBMSStorageEngine::savepoint_rollback(Session *thd, void *sv)

{

	PBMSStorageEngine * const hton = this;

#else

static int pbms_savepoint_rollback(handlerton *hton, THD *thd, void *sv)

{

#endif

	int			err = 0;

	CSThread	*self;

	PBMSResultRec result;

	if (pbms_enter_conn_no_thd(&self, &result))

		return 0;

	inner_();

	try_(a) {

		MSTransactionManager::rollbackTo(*((uint32_t*)sv));

	}

	catch_(a) {

		err = pbms_exception_to_result(&self->myException, &result);

	}

	cont_(a);

	return_(err);

}

#ifdef DRIZZLED

int PBMSStorageEngine::savepoint_release(Session *thd, void *sv)

{

	PBMSStorageEngine * const hton = this;

#else

static int pbms_savepoint_release(handlerton *hton, THD *thd, void *sv)

{

#endif

	return 0;

}

#ifdef DRIZZLED

void PBMSStorageEngine::drop_database(char *path)

{

	PBMSStorageEngine * const hton = this;

#else

static void pbms_drop_database(handlerton *hton, char *path)

{

#endif

	CSThread *self;

	char db_name[PATH_MAX];

	PBMSResultRec result;

	if (pbms_enter_conn_no_thd(&self, &result))

		return;

	inner_();

	cs_strcpy(PATH_MAX, db_name, cs_last_directory_of_path(path));

	cs_remove_dir_char(db_name);

	try_(a) {

		MSDatabase::dropDatabase(db_name);

	}

	catch_(a);

	self->logException();

	cont_(a);

	exit_();

}

static bool pbms_started = false;

#ifdef DRIZZLED

static int pbms_init_func(drizzled::plugin::Registry &registry)

#else

int pbms_discover_system_tables(handlerton *hton, THD* thd, const char *db, const char *name, uchar **frmblob, size_t *frmlen);

static int pbms_init_func(void *p)

#endif

{

	PBMSResultRec		result;

	int					err;

	int					my_res = 0;

	CSThread			*thread;

	ASSERT(!pbms_started);

	pbms_started = false;

	MSDatabase::gRepoThreshold = (uint64_t) cs_byte_size_to_int8(pbms_repository_threshold);

	MSDatabase::gTempLogThreshold = (uint64_t) cs_byte_size_to_int8(pbms_temp_log_threshold);

	{

		char info[120];

		snprintf(info, 120, "PrimeBase Media Stream (PBMS) Daemon %s loaded...", ms_version());

		CSL.logLine(NULL, CSLog::Protocol, info);

	}

	CSL.logLine(NULL, CSLog::Protocol, "Barry Leslie, PrimeBase Technologies GmbH, http://www.primebase.org");

	if ((err = MSEngine::startUp(&result))) {

		CSL.logLine(NULL, CSLog::Error, result.mr_message);

		return(1);

	}

#ifdef DRIZZLED

		pbms_hton= new PBMSStorageEngine(std::string("PBMS"));

		registry.add(pbms_hton);

#else

	pbms_hton = (handlerton *) p;

	pbms_hton->state = SHOW_OPTION_YES;

	pbms_hton->close_connection = pbms_close_connection; /* close_connection, cleanup thread related data. */

	pbms_hton->create = pbms_create_handler;

	pbms_hton->flags = HTON_CAN_RECREATE | HTON_HIDDEN;

	pbms_hton->drop_database = pbms_drop_database; /* Drop a database */

	pbms_hton->discover = pbms_discover_system_tables;

	pbms_hton->commit = pbms_commit; /* commit */

	pbms_hton->rollback = pbms_rollback; /* rollback */

	pbms_hton->savepoint_offset = 4;

	pbms_hton->savepoint_set = pbms_savepoint_set;

	pbms_hton->savepoint_rollback = pbms_savepoint_rollback; 

	pbms_hton->savepoint_release = pbms_savepoint_release; 

#endif

	/* Startup the Media Stream network: */

	cs_init_memory();

	CSThread::startUp();

	if (!(thread = CSThread::newCSThread())) {

		CSException::logOSError(CS_CONTEXT, ENOMEM);

		return(1);

	}

	if (!CSThread::attach(thread)) {

		thread->myException.log(NULL);

		CSThread::shutDown();

		cs_exit_memory();

		MSEngine::shutDown();

		return(1);

	}

	enter_();

	try_(a) {

		thread->threadName = CSString::newString("startup");

		//CSTest::runAll();

		MSDatabase::startUp(pbms_http_metadata_headers);

		MSTableList::startUp();

		MSSystemTableShare::startUp();

		MSNetwork::startUp(pbms_port);

		MSTransactionManager::startUp();

		MSNetwork::startNetwork();

	}

	catch_(a) {

		self->logException();

		my_res = 1;

	}

	cont_(a);

	if (my_res) {

		try_(b) {

			MSNetwork::shutDown();

			MSTransactionManager::shutDown();

			MSSystemTableShare::shutDown();

			MSDatabase::stopThreads();

			MSTableList::shutDown();

			MSDatabase::shutDown();

			CSThread::shutDown();

		}

		catch_(b) {

			self->logException();

		}

		cont_(b);

	}

	outer_();

	CSThread::detach(thread);

	if (my_res) {

		cs_exit_memory();

		MSEngine::shutDown();

	}

	else {

		srandom(time(NULL));

		pbms_started = true;

	}

	return(my_res);

}

#ifdef DRIZZLED

static int pbms_done_func(drizzled::plugin::Registry &registry)

#else

static int pbms_done_func(void *)

#endif

{

	CSThread	*thread;

	if (!pbms_started)

		return 0;

	CSL.logLine(NULL, CSLog::Protocol, "PrimeBase Media Stream (PBMS) Daemon shutdown...");

	/* Shutdown the Media Stream network. */

	if (!(thread = CSThread::newCSThread()))

		CSException::logOSError(CS_CONTEXT, ENOMEM);

	else if (!CSThread::attach(thread))

		thread->myException.log(NULL);

	else {

		enter_();

		try_(a) {

			thread->threadName = CSString::newString("shutdown");

			MSNetwork::shutDown();

			MSSystemTableShare::shutDown();

			/* Ensure that the database threads are stopped before

			 * freeing the tables.

			 */

			MSDatabase::stopThreads();

			MSTableList::shutDown();

			/* Databases must be shutdown after table because tables

			 * have references to repositories.

			 */

			MSDatabase::shutDown();

			/* Shutdown the transaction manager after the databases

			 * incase they want to commit or rollback a transaction.

			 */

			MSTransactionManager::shutDown();

		}

		catch_(a) {

			self->logException();

		}

		cont_(a);

		outer_();

		CSThread::shutDown();

		CSThread::detach(thread);

	}

	MSEngine::shutDown();

	cs_exit_memory();

	CSL.logLine(NULL, CSLog::Protocol, "PrimeBase Media Stream (PBMS) Daemon shutdown completed");

	pbms_started = false;

#ifdef DRIZZLED

	registry.remove(pbms_hton);

#endif

	return(0);

}

ha_pbms::ha_pbms(handlerton *hton, TABLE_SHARE *table_arg):

handler(hton, table_arg),

ha_open_tab(NULL),

ha_error(0)

{

	memset(&ha_result, 0, sizeof(PBMSResultRec));

}

MX_TABLE_TYPES_T ha_pbms::table_flags() const

{

	return (

		/* We need this flag because records are not packed

		 * into a table which means #ROWID != offset

		 */

		HA_REC_NOT_IN_SEQ |

		HA_CAN_SQL_HANDLER |

#if MYSQL_VERSION_ID > 50119

		/* We can do row logging, but not statement, because

		 * MVCC is not serializable!

		 */

		HA_BINLOG_ROW_CAPABLE |

#endif

		/*

		 * Auto-increment is allowed on a partial key.

		 */

		0);

}

int ha_pbms::open(const char *table_path, int mode, uint test_if_locked)

{

	CSThread *self;

	if ((ha_error = pbms_enter_conn(current_thd, &self, &ha_result)))

		return 1;

	inner_();

	try_(a) {

		ha_open_tab = MSSystemTableShare::openSystemTable(table_path, table);

		thr_lock_data_init(&ha_open_tab->myShare->myThrLock, &ha_lock, NULL);

		ref_length = ha_open_tab->getRefLen();

	}

	catch_(a) {

		ha_error = pbms_exception_to_result(&self->myException, &ha_result);

	}

	cont_(a);

	return_(ha_error != MS_OK);

}

int ha_pbms::close(void)

{

	CSThread *self;

	if ((ha_error = pbms_enter_conn(current_thd, &self, &ha_result)))

		return 1;

	inner_();

	if (ha_open_tab) {

		ha_open_tab->release();

		ha_open_tab = NULL;

	}

	outer_();

	pbms_exit_conn();

	return 0;

}

#ifdef PBMS_HAS_KEYS

/* Index access functions: */

int ha_pbms::index_init(uint idx, bool sorted __attribute__((unused)))

{

	int err = 0;

	enter_();

	try_(a) {

		ha_open_tab->index_init(idx);

	}

	catch_(a) {

		ha_error = pbms_exception_to_result(&self->myException, &ha_result);

		err = 1;

	}

	cont_(a);

	return_(err);

}

//-------

int ha_pbms::index_end()

{

	int err = 0;

	enter_();

	try_(a) {

		ha_open_tab->index_end();

	}

	catch_(a) {

		ha_error = pbms_exception_to_result(&self->myException, &ha_result);

		err = 1;

	}

	cont_(a);

	return_(err);

}

//-------

int ha_pbms::index_read(byte * buf, const byte * key,

							 uint key_len, enum ha_rkey_function find_flag)

{

	int err = 0;

	enter_();

	try_(a) {

		if (!ha_open_tab->index_read(buf, key, key_len, find_flag))

			err = HA_ERR_KEY_NOT_FOUND;

	}

	catch_(a) {

		ha_error = pbms_exception_to_result(&self->myException, &ha_result);

		err = 1;

	}

	cont_(a);

	return_(err);

}

//-------

int ha_pbms::index_read_idx(byte * buf, uint idx, const byte * key,

									 uint key_len, enum ha_rkey_function find_flag)

{

	int err = 0;

	enter_();

	try_(a) {

		if (!ha_open_tab->index_read_idx(buf, idx, key, key_len, find_flag))

			err = HA_ERR_KEY_NOT_FOUND;

	}

	catch_(a) {

		ha_error = pbms_exception_to_result(&self->myException, &ha_result);

		err = 1;

	}

	cont_(a);

	return_(err);

}

//-------

int ha_pbms::index_next(byte * buf)

{

	int err = 0;

	enter_();

	try_(a) {

		if (!ha_open_tab->index_next(buf))

			err = HA_ERR_END_OF_FILE;

	}

	catch_(a) {

		ha_error = pbms_exception_to_result(&self->myException, &ha_result);

		err = 1;

	}

	cont_(a);

	return_(err);

}

//-------

int ha_pbms::index_prev(byte * buf)