vxworks.cc

编译工具

//////////////////////////////////////////////////////////////////////////////
// Filename:	 vxWorks.cc
// Author:		Tihomir Sokcevic
//					Acterna, Eningen.
// Description: vxWorks adaptation of the omnithread wrapper classes
// Notes:		 Munching strategy is imperative
//////////////////////////////////////////////////////////////////////////////
// $Log: vxWorks.cc,v $
// Revision 1.1.2.6  2005/09/19 15:02:11  dgrisby
// Updated vxWorks omnithread options. Thanks Dirk Siebnich.
//
// Revision 1.1.2.5  2005/07/06 12:25:55  dgrisby
// Per thread data changes from April meant main thread's data was leaked.
//
// Revision 1.1.2.4  2005/04/25 18:16:24  dgrisby
// Always release per thread data in the thread it belongs to.
//
// Revision 1.1.2.3  2004/04/30 15:01:25  dgrisby
// Fix to dummy thread on vxWorks. Thanks Gary Block.
//
// Revision 1.1.2.2  2003/11/05 15:42:39  dgrisby
// vxWorks omnithread fixes from Jochen Gern.
//
// Revision 1.2  2003/11/05 11:05:38  gernjo
// In omni_condition, removed task synchronization via the waiters_ variable. Instead, in signal() re-take the semaphore after signaling. Changed stack size to 60000 due to enhanced stack requirements of release code.
//
// Revision 1.1.1.1  2003/10/20 10:15:50  gernjo
// Original distribution
//
// Revision 1.1.2.1  2003/02/17 02:03:11  dgrisby
// vxWorks port. (Thanks Michael Sturm / Acterna Eningen GmbH).
//
// Revision 1.1.1.1  2002/11/19 14:58:04  sokcevti
// OmniOrb4.0.0 VxWorks port
//
// Revision 1.4  2002/10/15 07:54:09  kuttlest
// change semaphore from SEM_FIFO to SEM_PRIO
// ---
//
// Revision 1.3  2002/07/05 07:38:52  engeln
// made priority redefinable on load time by defining int variables
// 	omni_thread_prio_low = 220;
// 	omni_thread_prio_normal = 110;
// 	omni_thread_prio_high = 55;
// the default priority is prio_normal.
// The normal priority default has been increased from 200 to 110 and the
//     high priority from 100 to 55.
// ---
//
// Revision 1.2  2002/06/14 12:44:57  engeln
// replaced possibly unsafe wakeup procedure in broadcast.
// ---
//
// Revision 1.1.1.1  2002/04/02 10:09:34  sokcevti
// omniORB4 initial realease
//
// Revision 1.0	2001/10/23 14:22:45	sokcevti
// Initial Version 4.00
// ---
//
//////////////////////////////////////////////////////////////////////////////


//////////////////////////////////////////////////////////////////////////////
// Include files
//////////////////////////////////////////////////////////////////////////////
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#include <time.h>
#include <omnithread.h>
#include <sysLib.h>

#include <assert.h>		// assert
#include <intLib.h>		// intContext


//////////////////////////////////////////////////////////////////////////////
// Local defines
//////////////////////////////////////////////////////////////////////////////
#define ERRNO(x) (((x) != 0) ? (errno) : 0)
#define THROW_ERRORS(x) { if((x) != OK) throw omni_thread_fatal(errno); }
#define OMNI_THREAD_ID	0x7F7155AAl
#define OMNI_STACK_SIZE 60000l

#ifdef _DEBUG
	#include <fstream>
	#define DBG_TRACE(X) X
#else // _DEBUG
	#define DBG_TRACE(X)
#endif // _DEBUG

#define DBG_ASSERT(X)

#define DBG_THROW(X) X

int omni_thread_prio_low = 220;
int omni_thread_prio_normal = 110;
int omni_thread_prio_high = 55;
///////////////////////////////////////////////////////////////////////////
//
// Mutex
//
///////////////////////////////////////////////////////////////////////////
omni_mutex::omni_mutex(void):m_bConstructed(false)
{
	mutexID = semMCreate(SEM_Q_PRIORITY | SEM_INVERSION_SAFE);

	DBG_ASSERT(assert(mutexID != NULL));

	if(mutexID==NULL)
	{
		DBG_TRACE(cout<<"Exception: omni_mutex::omni_mutex()  tid: "<<(int)taskIdSelf()<<endl);
		DBG_THROW(throw omni_thread_fatal(-1));
	}

	m_bConstructed = true;
}

omni_mutex::~omni_mutex(void)
{
	m_bConstructed = false;

	STATUS status = semDelete(mutexID);

	DBG_ASSERT(assert(status == OK));

	if(status != OK)
	{
		DBG_TRACE(cout<<"Exception: omni_mutex::~omni_mutex()  mutexID: "<<(int)mutexID<<" tid: "<<(int)taskIdSelf()<<endl);
		DBG_THROW(throw omni_thread_fatal(errno));
	}
}

/*
void omni_mutex::lock(void)
{
	DBG_ASSERT(assert(!intContext()));		// not in ISR context
	DBG_ASSERT(assert(m_bConstructed));

	STATUS status = semTake(mutexID, WAIT_FOREVER);

	DBG_ASSERT(assert(status == OK));

	if(status != OK)
	{
		DBG_TRACE(cout<<"Exception: omni_mutex::lock()  mutexID: "<<(int)mutexID<<" tid: "<<(int)taskIdSelf()<<endl);
		DBG_THROW(throw omni_thread_fatal(errno));
	}
}

void omni_mutex::unlock(void)
{
	DBG_ASSERT(assert(m_bConstructed));

	STATUS status = semGive(mutexID);

	DBG_ASSERT(assert(status == OK));

	if(status != OK)
	{
		DBG_TRACE(cout<<"Exception: omni_mutex::unlock()  mutexID: "<<(int)mutexID<<" tid: "<<(int)taskIdSelf()<<endl);
		DBG_THROW(throw omni_thread_fatal(errno));
	}
}
*/

///////////////////////////////////////////////////////////////////////////
//
// Condition variable
//
///////////////////////////////////////////////////////////////////////////
omni_condition::omni_condition(omni_mutex* m) : mutex(m)
{
	DBG_TRACE(cout<<"omni_condition::omni_condition  mutexID: "<<(int)mutex->mutexID<<" tid:"<<(int)taskIdSelf()<<endl);

	sema_ = semCCreate(SEM_Q_PRIORITY, 0);
	if(sema_ == NULL)
	{
		DBG_TRACE(cout<<"Exception: omni_condition::omni_condition()  tid: "<<(int)taskIdSelf()<<endl);
		DBG_THROW(throw omni_thread_fatal(errno));
	}
}

omni_condition::~omni_condition(void)
{
	STATUS status = semDelete(sema_);

	DBG_ASSERT(assert(status == OK));

	if(status != OK)
	{
		DBG_TRACE(cout<<"Exception: omni_condition::~omni_condition"<<endl);
		DBG_THROW(throw omni_thread_fatal(errno));
	}
}

void omni_condition::wait(void)
{
	STATUS status = OK;

	taskLock();	// disable task switch until semTake
				// else a broadcast may not wakeup
				// this waiting thread
	mutex->unlock();
		// release mutex
	status = semTake(sema_, WAIT_FOREVER);
		// wait at semaphore for a signal or broadcast
	taskUnlock();		// reenable task switch

	DBG_ASSERT(assert(status == OK));

	if(status != OK)
	{
		DBG_TRACE(cout<<"Exception: omni_condition::wait"<<endl);
		DBG_THROW(throw omni_thread_fatal(errno));
	}

	mutex->lock();
		// get owner of mutex before return
}


// The time given is absolute. Return 0 is timeout
int omni_condition::timedwait(unsigned long secs, unsigned long nanosecs)
{
	STATUS result = OK;
	timespec now;
	unsigned long timeout;
	int ticks;

	clock_gettime(CLOCK_REALTIME, &now);

	if(((unsigned long)secs <= (unsigned long)now.tv_sec) &&
		(((unsigned long)secs < (unsigned long)now.tv_sec) ||
		(nanosecs < (unsigned long)now.tv_nsec)))
		timeout = 0;
	else
		timeout = (secs-now.tv_sec) * 1000 + (nanosecs-now.tv_nsec) / 1000000l;

	// disable task lock to have an atomic unlock+semTake
	taskLock();

	// We keep the lock held just long enough to increment the count
	// of waiters by one.
	mutex->unlock();

	// Wait to be awakened by a signal() or broadcast().
	ticks = (timeout * sysClkRateGet()) / 1000L;
	result = semTake(sema_, ticks);

	// reenable task rescheduling
	taskUnlock();

	// We must always regain the <external_mutex>, even when errors
	// occur because that's the guarantee that we give to our callers.
	mutex->lock();

	if(result!= OK) // timeout
		return 0;

	return 1;
}

void omni_condition::signal(void)
{
	DBG_TRACE(cout<<"omni_condition::signal          mutexID: "<<(int)mutex->mutexID<<" tid:"<<(int)taskIdSelf()<<endl);

	STATUS status = semGive(sema_);

	DBG_ASSERT(assert(status == OK));

	if(status != OK)
	{
		DBG_TRACE(cout<<"Exception: omni_condition::signal"<<endl);
		DBG_THROW(throw omni_thread_fatal(errno));
	}
    
    // take unread message if no thread was waiting
    semTake(sema_, NO_WAIT);
}

void omni_condition::broadcast(void)
{
	DBG_TRACE(cout<<"omni_condition::broadcast       mutexID: "<<(int)mutex->mutexID<<" tid:"<<(int)taskIdSelf()<<endl);

	// Wake up all the waiters.
	STATUS status = semFlush(sema_);

	DBG_ASSERT(assert(status == OK));

	if(status != OK)
	{
		DBG_TRACE(cout<<"omni_condition::broadcast1! - thread:"<<omni_thread::self()->id()<<" SemID:"<<(int)sema_<<" errno:"<<errno<<endl);
		DBG_THROW(throw omni_thread_fatal(errno));
	}
}


///////////////////////////////////////////////////////////////////////////
//
// Counting semaphore
//
///////////////////////////////////////////////////////////////////////////
omni_semaphore::omni_semaphore(unsigned int initial)
{

	DBG_ASSERT(assert(0 <= (int)initial));		// POSIX expects only unsigned init values

	semID = semCCreate(SEM_Q_PRIORITY, (int)initial);

	DBG_ASSERT(assert(semID!=NULL));

	if(semID==NULL)
	{
		DBG_TRACE(cout<<"Exception: omni_semaphore::omni_semaphore"<<endl);
		DBG_THROW(throw omni_thread_fatal(-1));
	}
}

omni_semaphore::~omni_semaphore(void)
{
	STATUS status = semDelete(semID);

	DBG_ASSERT(assert(status == OK));

	if(status != OK)
	{
		DBG_TRACE(cout<<"Exception: omni_semaphore::~omni_semaphore"<<endl);
		DBG_THROW(throw omni_thread_fatal(errno));
	}
}

void omni_semaphore::wait(void)
{
	DBG_ASSERT(assert(!intContext()));		// no wait in ISR

	STATUS status = semTake(semID, WAIT_FOREVER);

	DBG_ASSERT(assert(status == OK));

	if(status != OK)
	{
		DBG_TRACE(cout<<"Exception: omni_semaphore::wait"<<endl);
		DBG_THROW(throw omni_thread_fatal(errno));
	}
}

int omni_semaphore::trywait(void)
{
	STATUS status = semTake(semID, NO_WAIT);

	DBG_ASSERT(assert(status == OK));

	if(status != OK)
	{
		if(errno == S_objLib_OBJ_UNAVAILABLE)
		{
			return 0;
		}
		else
		{
			DBG_ASSERT(assert(false));

			DBG_TRACE(cout<<"Exception: omni_semaphore::trywait"<<endl);
			DBG_THROW(throw omni_thread_fatal(errno));
		}
	}

	return 1;
}

void omni_semaphore::post(void)
{
	STATUS status = semGive(semID);

	DBG_ASSERT(assert(status == OK));

	if(status != OK)
	{
		DBG_TRACE(cout<<"Exception: omni_semaphore::post"<<endl);
		DBG_THROW(throw omni_thread_fatal(errno));
	}
}



///////////////////////////////////////////////////////////////////////////
//
// Thread
//
///////////////////////////////////////////////////////////////////////////


//
// static variables
//
omni_mutex* omni_thread::next_id_mutex = 0;
int omni_thread::next_id = 0;

// omniORB requires a larger stack size than the default (21120) on OSF/1
static size_t stack_size = OMNI_STACK_SIZE;


//
// Initialisation function (gets called before any user code).
//

static int& count() {
  static int the_count = 0;
  return the_count;
}

omni_thread::init_t::init_t(void)
{
	// Only do it once however many objects get created.
	if(count()++ != 0)
		return;

	attach();
}

omni_thread::init_t::~init_t(void)
{
    if (--count() != 0) return;

    omni_thread* self = omni_thread::self();
    if (!self) return;

    taskTcb(taskIdSelf())->spare1 = 0;
    delete self;

    delete next_id_mutex;
}


//
// Wrapper for thread creation.
//
extern "C" void omni_thread_wrapper(void* ptr)
{
	omni_thread* me = (omni_thread*)ptr;

	DBG_TRACE(cout<<"omni_thread_wrapper: thread "<<me->id()<<" started\n");

	//
	// We can now tweaked the task info since the tcb exist now
	//
	me->mutex.lock();	// To ensure that start has had time to finish
	taskTcb(me->tid)->spare1 = OMNI_THREAD_ID;
	taskTcb(me->tid)->spare2 = (int)ptr;
	me->mutex.unlock();

	//
	// Now invoke the thread function with the given argument.
	//
	if(me->fn_void != NULL)
	{
		(*me->fn_void)(me->thread_arg);
		omni_thread::exit();
	}

	if(me->fn_ret != NULL)
	{
		void* return_value = (*me->fn_ret)(me->thread_arg);
		omni_thread::exit(return_value);
	}

	if(me->detached)
	{
		me->run(me->thread_arg);
		omni_thread::exit();
	}
	else
	{
		void* return_value = me->run_undetached(me->thread_arg);
		omni_thread::exit(return_value);
	}
}


//
// Special functions for VxWorks only
//
void omni_thread::attach(void)
{
	DBG_TRACE(cout<<"omni_thread_attach: VxWorks mapping thread initialising\n");

	int _tid = taskIdSelf();

	// Check the task is not already attached
	if(taskTcb(_tid)->spare1 == OMNI_THREAD_ID)
		return;

	// Create the mutex required to lock the threads debugging id (create before the thread!!!)
	if(next_id_mutex == 0)
		next_id_mutex = new omni_mutex;

	// Create a thread object for THIS running process
	omni_thread* t = new omni_thread;

	// Lock its mutex straigh away!