~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 (H&G2JCtL)

 * Continued development: Barry Leslie

 *

 * 2007-05-20

 *

 * CORE SYSTEM:

 * A independently running thread.

 *

 */

#ifndef __CSTHREAD_H__

#define __CSTHREAD_H__

#include <pthread.h>

#include <setjmp.h>

#include "CSDefs.h"

#include "CSMutex.h"

#include "CSException.h"

#include "CSStorage.h"

using namespace std;

#define CS_THREAD_TYPE				int

/* Types of threads: */

#define CS_ANY_THREAD				0

#define CS_THREAD					1

typedef struct CSCallStack {

	const char*	cs_func;

	const char*	cs_file;

	int			cs_line;

} CSCallStack, *CSCallStackPtr;

/* 

 * The release stack contains objects that need to be

 * released when an exception occurs.

 */

#define CS_RELEASE_OBJECT		1

#define CS_RELEASE_MUTEX		2

#define CS_RELEASE_POOLED		3

#define CS_RELEASE_MEM			4

typedef struct CSRelease {

	int						r_type;

	union {

		CSObject			*r_object;					/* The object to be released. */

		CSMutex				*r_mutex;					/* The mutex to be unlocked! */

		CSPooled			*r_pooled;

		void				*r_mem;

	} x;

} CSReleaseRec, *CSReleasePtr;

typedef struct CSJumpBuf {

	CSReleasePtr			jb_res_top;

	int						jb_call_top;

	jmp_buf					jb_buffer;

} CSJumpBufRec, *CSJumpBufPtr;

class CSThreadList: public CSLinkedList, public CSMutex {

public:

	CSThreadList():

		CSLinkedList(),

		CSMutex()

	{

	}

	virtual ~CSThreadList() {

		stopAllThreads();

	}

	/**

	 * Send the given signal to all threads, except to self!

	 */

	void signalAllThreads(int sig);

	void quitAllThreads();

	void stopAllThreads();

};

typedef void *(*ThreadRunFunc)();

class CSThread : public CSRefObject {

public:

	/* The name of the thread. */

	CSString		*threadName;

	CSThreadList	*myThreadList;				/* The thread list that this thread belongs to. */

	/* If this value is non-zero, this signal is pending and

	 * must be thrown.

	 *

	 * SIGTERM, SIGQUIT - Means the thread has been terminated.

	 * SIGINT - Means the thread has been interrupted.

	 *

	 * When a signal is throw it clears this value. This includes

	 * the case when system calls return error due to interrupt.

	 */

	int				signalPending;

	bool			ignoreSignals;

	/* Set to true once the thread is running (never reset!). */

	bool			isRunning;

	/* Set to true when the thread must quit (never reset!): */

	bool			myMustQuit;	

	/* Set to true when this tread was initialized through the internal PBMS api. */

	/* When this is the case than it must only be freed via the API as well. */

	bool			pbms_api_owner;

	CSException		myException;

	/* Transaction references. */

#ifdef DRIZZLED

	CSSortedList	mySavePoints;

#endif

	uint32_t			myTID;			// Current transaction ID

	uint32_t			myTransRef;		// Reference to the current transaction cache index

	bool			myIsAutoCommit;	// Is the current transaction in auto commit mode.

	uint32_t			myCacheVersion;	// The last transaction cache version checked. Used during overflow.

	bool			myStartTxn;		// A flag to indicate the start of a new transaction.

	uint32_t			myStmtCount;	// Counts the number of statements in the current transaction.

	uint32_t			myStartStmt;	// The myStmtCount at the start of the last logical statement. (An update is 2 statements but only 1 logical statement.)

	void			*myInfo;		// A place to hang some info. (Be carefull with this!)

	/* The call stack */

	int				callTop;

	CSCallStack		callStack[CS_CALL_STACK_SIZE];

	/* The long jump stack: */

	int				jumpDepth;							/* The current jump depth */

	CSJumpBufRec	jumpEnv[CS_JUMP_STACK_SIZE];		/* The process environment to be restored on exception */

	/* The release stack */

 	CSReleasePtr	relTop;								/* The top of the resource stack (reference next free space). */

	CSReleaseRec	relStack[CS_RELEASE_STACK_SIZE];	/* Temporary data to be freed if an exception occurs. */

	CSThread(CSThreadList *list):

		CSRefObject(),

		threadName(NULL),

		myThreadList(list),

		signalPending(0),

		ignoreSignals(false),

		isRunning(false),

		myMustQuit(false),

		pbms_api_owner(false),

		myTID(0),

		myTransRef(0),

		myIsAutoCommit(true),

		myCacheVersion(0),

		myStartTxn(true),

		myStmtCount(0),

		myStartStmt(0),

		myInfo(NULL),

		callTop(0),

		jumpDepth(0),

		relTop(relStack),

		iIsMain(false),

		iRunFunc(NULL),

		iNextLink(NULL),

		iPrevLink(NULL)

	{

	}

	virtual ~CSThread() {

		if (threadName)

			threadName->release();

	}

    /**

     * Task to be performed by this thread.

	 *

     * @exception CSSystemException thrown if thread cannot be run.

	 */

	virtual void *run();

	/**

	 * Start execution of the thread.

	 *

     * @exception CSSystemException thrown if thread is invalid.

	 */

	void start();

	/*

	 * Stop execution of the thread.

	 */

	virtual void stop();

	/**

	 * Wait for this thread to die.

	 *

     * @exception CSSystemException thrown if thread is invalid.

	 */

	void *join();

	/**

	 * Signal the thread. Throws CSSystemException 

     * if the thread is invalid.

	 */

	void signal(unsigned int);

	void setSignalPending(unsigned int);

	/**

	 * Check to see if we have been interrupted by a signal

	 * (i.e. there is a signal pending).

	 * This function throws CSSignalException if

	 * there is a signal pending.

	 */

	void interrupted() { if (signalPending) throwSignal(); }

	void throwSignal();

	/* Log the stack to the specified depth along with the message. */

	void logStack(int depth, const char *msg);

	/* Log the exception, and the current stack. */

	void logException();

	/* Log the exception, and the current stack. */

	void logMessage();

	/*

	 * Return true if this is the main thread.

	 */

	bool isMain();

	/*

	 * Throwing exceptions:

	 */

	void releaseObjects(CSReleasePtr top);

	void throwException();

	void caught();

	bool isMe(CSThread *me) { return (me->iThread == iThread);}

	/* Make this object linkable: */

	virtual CSObject *getNextLink() { return iNextLink; }

	virtual CSObject *getPrevLink() { return iPrevLink; }

	virtual void setNextLink(CSObject *link) { iNextLink = link; }

	virtual void setPrevLink(CSObject *link) { iPrevLink = link; }

	friend class CSDaemon;

private:

	pthread_t		iThread;

	bool			iIsMain;

	ThreadRunFunc	iRunFunc;

	CSObject		*iNextLink;

	CSObject		*iPrevLink;

	void addToList();

	void removeFromList();

public:

	/* Each thread stores is thread object in this key: */

	static pthread_key_t sThreadKey;

   /**

     * Put the currently executing thread to sleep for a given amount of

     * time.

     *

     * @param timeout maximum amount of time (milliseconds) this method could block

     *

     * @exception TDInterruptedException thrown if the threads sleep is interrupted

     *            before <i>timeout</i> milliseconds expire.

     */

	static void sleep(unsigned long timeout);

	/* Do static initialization and de-initialization. */

	static bool isUp;

	static bool startUp();

	static void shutDown();

	/* Attach and detach an already running thread: */

	static bool attach(CSThread *thread);

	static void detach(CSThread *thread);

	/**

	 * Return the thread object of the current

	 * thread.

	 */

	static CSThread *getSelf();

	static bool setSelf(CSThread *self);

	static CSThread *newCSThread();

	static CSThread *newThread(CSString *name, ThreadRunFunc run_func, CSThreadList *list);

	/* called for a newly created thread. */

	static void *dispatch(void *arg);

};

class CSDaemon : public CSThread, public CSSync {

public:

	time_t			myWaitTime;					/* Wait time in milli-seconds */

	CSDaemon(time_t wait_time, CSThreadList *list);

	CSDaemon(CSThreadList *list);

	virtual ~CSDaemon() { }

	virtual void *run();

	virtual bool initializeWork() { return true; };

	virtual bool doWork();

	virtual void *completeWork() { return NULL; };

	virtual bool handleException();

	virtual void stop();

	void wakeup();

	void suspend();

	bool isSuspend() { return (iSuspendCount != 0);} // Don't use iSuspended, we are interested in if suspend() was called.

	void resume();

	virtual void returnToPool() {

		resume();

		release();

	}

	void suspended();

	void suspendedWait();

	void suspendedWait(time_t milli_sec);

private:

	bool			iSuspended;

	uint32_t			iSuspendCount;

};

#endif