pegasus/src/Pegasus/Common/Thread.cpp - diff

Return to Thread.cpp CVS log

Up to [Pegasus] / pegasus / src / Pegasus / Common

Diff for /pegasus/src/Pegasus/Common/Thread.cpp between version 1.80 and 1.81

version 1.80, 2005/03/22 00:36:02

version 1.81, 2005/04/01 17:26:52

Line 188

PEG_METHOD_ENTER(TRC_THREAD, "Thread::setCurrent");

if (Thread::initializeKey() == 0)

{

if (pegasus_set_thread_specific(Thread::_platform_thread_key,

if (pegasus_set_thread_specific(

(void *) thrd) == 0)

Thread::_platform_thread_key, (void *) thrd) == 0)

{

Tracer::trace(TRC_THREAD, Tracer::LEVEL4,

"Successful set Thread * into thread specific storage");

Line 197

else

{

Tracer::trace(TRC_THREAD, Tracer::LEVEL4,

"ERROR: got error setting Thread * into thread specific storage");

"ERROR: error setting Thread * into thread specific storage");

}

PEG_METHOD_EXIT();

Line 250

// l10n end

class auto_int

///////////////////////////////////////////////////////////////////////////////

{

public:

// ThreadPool

auto_int(AtomicInt* num)

: _int(num)

///////////////////////////////////////////////////////////////////////////////

{

_int->operator++();

}

~auto_int(void)

{

_int->operator--();

}

AtomicInt *_int;

};

AtomicInt _idle_control;

DQueue<ThreadPool> ThreadPool::_pools(true);

void ThreadPool::kill_idle_threads(void)

ThreadPool::ThreadPool(

Sint16 initialSize,

const char* key,

Sint16 minThreads,

Sint16 maxThreads,

struct timeval& deallocateWait)

: _maxThreads(maxThreads),

_minThreads(minThreads),

_currentThreads(0),

_idleThreads(true),

_runningThreads(true),

_dying(0)

{

static struct timeval now, last = {0, 0};

_deallocateWait.tv_sec = deallocateWait.tv_sec;

_deallocateWait.tv_usec = deallocateWait.tv_usec;

pegasus_gettimeofday(&now);

memset(_key, 0x00, 17);

if(now.tv_sec - last.tv_sec > 5)

if (key != 0)

{

_pools.lock();

ThreadPool *p = _pools.next(0);

while(p != 0)

{

try

{

p->kill_dead_threads();

strncpy(_key, key, 16);

}

catch(...)

if ((_maxThreads > 0) && (_maxThreads < initialSize))

{

_maxThreads = initialSize;

}

p = _pools.next(p);

}

_pools.unlock();

pegasus_gettimeofday(&last);

}

ThreadPool::ThreadPool(Sint16 initial_size,

if (_minThreads > initialSize)

const char *key,

Sint16 min,

Sint16 max,

struct timeval & alloc_wait,

struct timeval & dealloc_wait)

: _max_threads(max), _min_threads(min),

_current_threads(0),

_pool(true), _running(true),

_dying(0)

{

_allocate_wait.tv_sec = alloc_wait.tv_sec;

_minThreads = initialSize;

_allocate_wait.tv_usec = alloc_wait.tv_usec;

}

_deallocate_wait.tv_sec = dealloc_wait.tv_sec;

_deallocate_wait.tv_usec = dealloc_wait.tv_usec;

memset(_key, 0x00, 17);

if(key != 0)

strncpy(_key, key, 16);

if(_max_threads > 0 && _max_threads < initial_size)

_max_threads = initial_size;

if(_min_threads > initial_size)

_min_threads = initial_size;

int i;

for (int i = 0; i < initialSize; i++)

for(i = 0; i < initial_size; i++)

{

_link_pool(_init_thread());

_addToIdleThreadsQueue(_initializeThread());

}

_pools.insert_last(this);

}

ThreadPool::~ThreadPool()

// Note: <<< Fri Oct 17 09:19:03 2003 mdd >>>

// the pegasus_yield() calls that preceed each th->join() are to

// give a thread on the running list a chance to reach a cancellation

// point before the join

ThreadPool::~ThreadPool(void)

{

PEG_METHOD_ENTER(TRC_THREAD, "Thread::~ThreadPool");

PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::~ThreadPool");

try

{

// Set the dying flag so all thread know the destructor has been entered

_dying++;

// remove from the global pools list

while (_currentThreads.value() > 0)

_pools.remove(this);

// start with idle threads.

Thread *th = 0;

th = _pool.remove_first();

Semaphore* sleep_sem;

while(th != 0)

{

sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");

Thread* thread = _idleThreads.remove_first();

PEGASUS_ASSERT(sleep_sem != 0);

if (thread != 0)

if(sleep_sem == 0)

{

th->dereference_tsd();

_cleanupThread(thread);

_currentThreads--;

}

else

{

// Signal to get the thread out of the work loop.

sleep_sem->signal();

// Signal to get the thread past the end. See the comment

// "wait to be awakend by the thread pool destructor"

// Note: the current implementation of Thread for Windows

// does not implement "pthread" cancelation points so this

// is needed.

sleep_sem->signal();

th->dereference_tsd();

th->join();

delete th;

}

th = _pool.remove_first();

}

while(_idle_control.value())

pegasus_yield();

{

th = _running.remove_first();

while(th != 0)

{

// signal the thread's sleep semaphore

sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");

PEGASUS_ASSERT(sleep_sem != 0);

if(sleep_sem == 0 )

{

th->dereference_tsd();

}

else

{

sleep_sem->signal();

th->dereference_tsd();

//th->cancel();

pegasus_yield();

th->join();

delete th;

}

th = _running.remove_first();

}

catch(...)

{

}

// make this static to the class

PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL ThreadPool::_loop(void *parm)

{

PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::_loop");

Thread *myself = (Thread *)parm;

if(myself == 0)

PEGASUS_ASSERT(myself != 0);

{

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"ThreadPool::_loop: Thread pointer is null");

PEG_METHOD_EXIT();

throw NullPointer();

}

// l10n

// Set myself into thread specific storage

// This will allow code to get its own Thread

Thread::setCurrent(myself);

ThreadPool *pool = (ThreadPool *)myself->get_parm();

if(pool == 0 )

PEGASUS_ASSERT(pool != 0);

{

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"ThreadPool::_loop: ThreadPool pointer is null");

PEG_METHOD_EXIT();

throw NullPointer();

}

if(pool->_dying.value())

{

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"ThreadPool::_loop: ThreadPool is dying(1)");

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

}

Semaphore *sleep_sem = 0;

Semaphore *blocking_sem = 0;

struct timeval* lastActivityTime = 0;

struct timeval *deadlock_timer = 0;

try

{

sleep_sem = (Semaphore *)myself->reference_tsd("sleep sem");

myself->dereference_tsd();

deadlock_timer = (struct timeval *)myself->reference_tsd("deadlock timer");

PEGASUS_ASSERT(sleep_sem != 0);

lastActivityTime =

(struct timeval *)myself->reference_tsd("last activity time");

myself->dereference_tsd();

PEGASUS_ASSERT(lastActivityTime != 0);

}

catch(...)

{

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"ThreadPool::_loop: Failure getting sleep_sem or deadlock_timer.");

"ThreadPool::_loop: Failure getting sleep_sem or "

_graveyard(myself);

"lastActivityTime.");

PEGASUS_ASSERT(false);

pool->_idleThreads.remove(myself);

pool->_currentThreads--;

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

return((PEGASUS_THREAD_RETURN)1);

}

if(sleep_sem == 0 || deadlock_timer == 0)

{

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"ThreadPool::_loop: sleep_sem or deadlock_timer are null.");

_graveyard(myself);

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

}

while(1)

{

if(pool->_dying.value())

break;

try

{

Boolean ignoreInterrupt = false;

sleep_sem->wait();

sleep_sem->wait(ignoreInterrupt);

}

catch (WaitInterrupted &e)

{

/* From the sem_wait manpage:

The sem_trywait() and sem_wait() functions may fail if:

EINTR A signal interrupted this function.

PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"Sleep semaphore wait failed. Doing a continue");

continue;

}

catch(IPCException& )

catch (...)

{

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"ThreadPool::_loop: failure on sleep_sem->wait().");

_graveyard(myself);

PEGASUS_ASSERT(false);

pool->_idleThreads.remove(myself);

pool->_currentThreads--;

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

return((PEGASUS_THREAD_RETURN)1);

}

// when we awaken we reside on the running queue, not the pool queue

// When we awaken we reside on the _runningThreads queue, not the

// _idleThreads queue.

PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *_work)(void *) = 0;

PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL* work)(void *) = 0;

void *parm = 0;

Semaphore* blocking_sem = 0;

try

{

_work = (PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *)) \

work = (PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *))

myself->reference_tsd("work func");

myself->dereference_tsd();

parm = myself->reference_tsd("work parm");

myself->dereference_tsd();

blocking_sem = (Semaphore *)myself->reference_tsd("blocking sem");

myself->dereference_tsd();

}

catch(IPCException &)

catch (...)

{

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"ThreadPool::_loop: Failure accessing work func, work parm, or blocking sem.");

"ThreadPool::_loop: Failure accessing work func, work parm, "

"or blocking sem.");

* We cannot move ourselves to the dead queue b/c the TSD might be still

PEGASUS_ASSERT(false);

* locked and _graveyard is not equipped to de-lock (dereference_tsd) the TSD.

pool->_idleThreads.remove(myself);

* Only the kill_dead_threads has enough logic to handle such situations.

pool->_currentThreads--;

_graveyard( myself);

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

return((PEGASUS_THREAD_RETURN)1);

}

if(_work == 0)

if (work == 0)

{

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"ThreadPool::_loop: work func is null.");

"ThreadPool::_loop: work func is 0, meaning we should exit.");

PEG_METHOD_EXIT();

break;

return((PEGASUS_THREAD_RETURN)0);

}

if(_work ==

(PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *)) &_undertaker)

{

* The undertaker is set by ThreadPool::kill_dead_threads which awakens this thread,

* joins it and then removes it from the queue. Hence no reason to go to the

_graveyard( myself);

PEG_METHOD_EXIT();

_work(parm);

}

gettimeofday(deadlock_timer, NULL);

gettimeofday(lastActivityTime, NULL);

if (pool->_dying.value() == 0)

{

try

{

PEG_TRACE_STRING(TRC_THREAD, Tracer::LEVEL4,

PEG_TRACE_STRING(TRC_THREAD, Tracer::LEVEL4, "Work starting.");

"Worker started");

work(parm);

_work(parm);

PEG_TRACE_STRING(TRC_THREAD, Tracer::LEVEL4, "Work finished.");

PEG_TRACE_STRING(TRC_THREAD, Tracer::LEVEL4,

"Worker finished");

}

catch(Exception & e)

{

PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,

String("Exception from _work in ThreadPool::_loop: ") +

String("Exception from work in ThreadPool::_loop: ") +

e.getMessage());

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

}

#if !defined(PEGASUS_OS_LSB)

catch (exception& e)

{

PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,

String("Exception from _work in ThreadPool::_loop: ") +

String("Exception from work in ThreadPool::_loop: ") +

e.what());

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

}

#endif

catch(...)

{

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"ThreadPool::_loop: execution of _work failed.");

"Unknown exception from work in ThreadPool::_loop.");

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

}

// put myself back onto the available list

try

{

if(pool->_dying.value() == 0)

gettimeofday(lastActivityTime, NULL);

{

gettimeofday(deadlock_timer, NULL);

if( blocking_sem != 0 )

blocking_sem->signal();

// If we are not on _running then ~ThreadPool has removed

// us and now "owns" our pointer.

if ( pool->_running.remove((void *)myself) != 0 )

{

pool->_pool.insert_first(myself);

}

else

{

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"ThreadPool::_loop: Failed to remove thread from running queue.");

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

}

else

{

PEG_METHOD_EXIT();

blocking_sem->signal();

return((PEGASUS_THREAD_RETURN)0);

}

Boolean removed = pool->_runningThreads.remove((void *)myself);

PEGASUS_ASSERT(removed);

pool->_idleThreads.insert_first(myself);

}

catch(...)

{

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"ThreadPool::_loop: Adding thread to idle pool failed.");

PEGASUS_ASSERT(false);

pool->_currentThreads--;

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

return((PEGASUS_THREAD_RETURN)1);

}

// TODO: Why is this needed? Why not just continue?

// wait to be awakend by the thread pool destructor

//sleep_sem->wait();

myself->test_cancel();

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

}

Boolean ThreadPool::allocate_and_awaken(void *parm,

Boolean ThreadPool::allocate_and_awaken(

PEGASUS_THREAD_RETURN \

void* parm,

(PEGASUS_THREAD_CDECL *work)(void *),

PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL* work)(void *),

Semaphore *blocking)

throw(IPCException)

{

PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::allocate_and_awaken");

Line 672

Line 495

gettimeofday(&start, NULL);

Thread *th = 0;

th = _pool.remove_first();

th = _idleThreads.remove_first();

if (th == 0)

{

if(_max_threads == 0 || _current_threads < _max_threads)

if ((_maxThreads == 0) || (_currentThreads < _maxThreads))

{

th = _init_thread();

th = _initializeThread();

}

Line 693

Line 516

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"ThreadPool::allocate_and_awaken: Insufficient resources: "

" pool = %s, running threads = %d, idle threads = %d",

_key, _running.count(), _pool.count());

_key, _runningThreads.count(), _idleThreads.count());

return false;

}

Line 713

Line 536

th->put_tsd("blocking sem", NULL, sizeof(Semaphore *), blocking);

// put the thread on the running list

_running.insert_first(th);

_runningThreads.insert_first(th);

// signal the thread's sleep semaphore to awaken it

Semaphore *sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");

PEGASUS_ASSERT(sleep_sem != 0);

if(sleep_sem == 0)

{

th->dereference_tsd();

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"ThreadPool::allocate_and_awaken: thread data is corrupted.");

PEG_METHOD_EXIT();

throw NullPointer();

}

Tracer::trace(TRC_THREAD, Tracer::LEVEL4, "Signal thread to awaken");

sleep_sem->signal();

th->dereference_tsd();

Line 743

Line 559

}

// caller is responsible for only calling this routine during slack periods

// but should call it at least once per _deallocate_wait for the pool q

// but should call it at least once per _deallocateWait interval.

Uint32 ThreadPool::kill_dead_threads(void)

Uint32 ThreadPool::cleanupIdleThreads()

throw(IPCException)

{

// Since the kill_dead_threads, ThreadPool or allocate_and_awaken

PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::cleanupIdleThreads");

// manipulate the threads on the ThreadPool queues, they should never

// be allowed to run at the same time.

PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::kill_dead_threads");

Uint32 numThreadsCleanedUp = 0;

// << Thu Oct 23 14:41:02 2003 mdd >>

// not true, the queues are thread safe. they are syncrhonized.

auto_int do_not_destruct(&_idle_control);

Uint32 numIdleThreads = _idleThreads.count();

for (Uint32 i = 0; i < numIdleThreads; i++)

try

{

struct timeval now;

// Do not dip below the minimum thread count

gettimeofday(&now, NULL);

if (_currentThreads.value() <= (Uint32)_minThreads)

Uint32 bodies = 0;

if (_dying.value())

{

return 0;

break;

}

Thread *th = 0;

Thread* thread = _idleThreads.remove_last();

internal_dq idq;

if(_pool.count() > 0 )

// If there are no more threads in the _idleThreads queue, we're done.

{

if (thread == 0)

try

{

_pool.try_lock();

break;

}

catch(...)

{

return bodies;

}

struct timeval dt = { 0, 0 };

struct timeval* lastActivityTime;

struct timeval *dtp;

th = _pool.next(th);

while (th != 0 )

{

try

{

dtp = (struct timeval *)th->try_reference_tsd("deadlock timer");

lastActivityTime = (struct timeval *)thread->try_reference_tsd(

"last activity time");

PEGASUS_ASSERT(lastActivityTime != 0);

}

catch(...)

{

_pool.unlock();

PEGASUS_ASSERT(false);

return bodies;

_idleThreads.insert_last(thread);

break;

}

if(dtp != 0)

Boolean cleanupThisThread =

{

_timeIntervalExpired(lastActivityTime, &_deallocateWait);

memcpy(&dt, dtp, sizeof(struct timeval));

thread->dereference_tsd();

if (cleanupThisThread)

{

_cleanupThread(thread);

_currentThreads--;

numThreadsCleanedUp++;

}

th->dereference_tsd();

else

struct timeval deadlock_timeout;

Boolean too_long;

too_long = check_time(&dt, get_deallocate_wait(&deadlock_timeout));

if( true == too_long)

{

// escape if we are down to the minimum thread count

_current_threads--;

if( _current_threads.value() < (Uint32)_min_threads )

{

_current_threads++;

_idleThreads.insert_first(thread);

th = _pool.next(th);

continue;

}

th = _pool.remove_no_lock((void *)th);

idq.insert_first((void*)th);

}

th = _pool.next(th);

}

_pool.unlock();

PEG_METHOD_EXIT();

return numThreadsCleanedUp;

}

th = (Thread*)idq.remove_last();

void ThreadPool::_cleanupThread(Thread* thread)

while(th != 0)

{

th->delete_tsd("work func");

PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::cleanupThread");

th->put_tsd("work func", NULL,

// Set the "work func" and "work parm" to 0 so _loop() knows to exit.

thread->delete_tsd("work func");

thread->put_tsd(

"work func", 0,

sizeof( PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *)),

(void *)&_undertaker);

(void *) 0);

th->delete_tsd("work parm");

thread->delete_tsd("work parm");

th->put_tsd("work parm", NULL, sizeof(void *), th);

thread->put_tsd("work parm", 0, sizeof(void *), 0);

// signal the thread's sleep semaphore to awaken it

Semaphore *sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");

Semaphore* sleep_sem = (Semaphore *)thread->reference_tsd("sleep sem");

PEGASUS_ASSERT(sleep_sem != 0);

bodies++;

th->dereference_tsd();

sleep_sem->signal();

th->join(); // Clean up the thread

thread->dereference_tsd();

delete th;

th = (Thread*)idq.remove_last();

thread->join();

}

delete thread;

return bodies;

}

catch (...)

{

}

PEG_METHOD_EXIT();

return 0;

}

Boolean ThreadPool::_timeIntervalExpired(

Boolean ThreadPool::check_time(struct timeval *start, struct timeval *interval)

struct timeval* start,

struct timeval* interval)

{

// never time out if the interval is zero

if(interval && interval->tv_sec == 0 && interval->tv_usec == 0)

if (interval && (interval->tv_sec == 0) && (interval->tv_usec == 0))

{

return false;

}

struct timeval now , finish , remaining ;

Uint32 usec;

pegasus_gettimeofday(&now);

/* remove valgrind error */

pegasus_gettimeofday(&remaining); // Avoid valgrind error

pegasus_gettimeofday(&remaining);

finish.tv_sec = start->tv_sec + interval->tv_sec;

usec = start->tv_usec + interval->tv_usec;

Line 878

Line 664

usec %= 1000000;

finish.tv_usec = usec;

if ( timeval_subtract(&remaining, &finish, &now) )

return (timeval_subtract(&remaining, &finish, &now) != 0);

return true;

else

return false;

}

PEGASUS_THREAD_RETURN ThreadPool::_undertaker( void *parm )

void ThreadPool::_deleteSemaphore(void *p)

{

PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::_undertaker");

exit_thread((PEGASUS_THREAD_RETURN)1);

PEG_METHOD_EXIT();

return (PEGASUS_THREAD_RETURN)1;

}

PEGASUS_THREAD_RETURN ThreadPool::_graveyard(Thread *t)

{

PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::_graveyard");

ThreadPool *pool = (ThreadPool *)t->get_parm();

if(pool == 0 ) {

Tracer::trace(TRC_THREAD, Tracer::LEVEL2,

"Could not obtain the pool information from the Thread.", t);

return (PEGASUS_THREAD_RETURN)1;

}

if (pool->_pool.exists(t))

{

if (pool->_pool.remove( (void *) t) != 0)

{

Tracer::trace(TRC_THREAD, Tracer::LEVEL4,

"Moving thread %p", t);

/* We are moving the thread to the _running queue b/c

_only_ kill_dead_threads has enough logic to take care

of cleaning up the threads.*/

pool->_running.insert_first( t );

}

else

{

Tracer::trace(TRC_THREAD, Tracer::LEVEL4,

"Could not move Thread %p from _pool to _runing queue.", t);

return (PEGASUS_THREAD_RETURN)1;

}

else if (pool->_running.exists(t))

{

Tracer::trace(TRC_THREAD, Tracer::LEVEL4,

"Thread %p is on _running queue. Letting kill_dead_threads take care of the problem.", t);

return (PEGASUS_THREAD_RETURN)1;

}

else

{

Tracer::trace(TRC_THREAD, Tracer::LEVEL2,

"Thread is not on any queue! Moving it to the running queue.");

pool->_running.insert_first( t );

}

PEG_METHOD_EXIT();

return (PEGASUS_THREAD_RETURN)0;

}

void ThreadPool::_sleep_sem_del(void *p)

{

if(p != 0)

{

delete (Semaphore *)p;

}

Boolean ThreadPool::_check_dealloc(struct timeval *start)

Thread* ThreadPool::_initializeThread()

{

return(check_time(start, &_deallocate_wait));

PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::_initializeThread");

}

Thread *ThreadPool::_init_thread(void) throw(IPCException)

{

PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::_init_thread");

Thread *th = (Thread *) new Thread(_loop, this, false);

// allocate a sleep semaphore and pass it in the thread context

// initial count is zero, loop function will sleep until

// we signal the semaphore

Semaphore *sleep_sem = (Semaphore *) new Semaphore(0);

th->put_tsd("sleep sem", &_sleep_sem_del, sizeof(Semaphore), (void *)sleep_sem);

th->put_tsd(

"sleep sem", &_deleteSemaphore, sizeof(Semaphore), (void *)sleep_sem);

struct timeval *dldt = (struct timeval *) ::operator new(sizeof(struct timeval));

pegasus_gettimeofday(dldt);

th->put_tsd("deadlock timer", thread_data::default_delete, sizeof(struct timeval), (void *)dldt);

struct timeval* lastActivityTime =

// thread will enter _loop(void *) and sleep on sleep_sem until we signal it

(struct timeval *) ::operator new(sizeof(struct timeval));

pegasus_gettimeofday(lastActivityTime);

th->put_tsd("last activity time", thread_data::default_delete,

sizeof(struct timeval), (void *)lastActivityTime);

// thread will enter _loop() and sleep on sleep_sem until we signal it

if (!th->run())

{

delete th;

return 0;

}

_current_threads++;

_currentThreads++;

pegasus_yield();

PEG_METHOD_EXIT();

return th;

}

void ThreadPool::_link_pool(Thread *th) throw(IPCException)

void ThreadPool::_addToIdleThreadsQueue(Thread* th)

{

if(th == 0)

{

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"ThreadPool::_link_pool: Thread pointer is null.");

"ThreadPool::_addToIdleThreadsQueue: Thread pointer is null.");

throw NullPointer();

}

try

{

_pool.insert_first(th);

_idleThreads.insert_first(th);

}

catch(...)

{

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"ThreadPool::_link_pool: _pool.insert_first failed.");

"ThreadPool::_addToIdleThreadsQueue: _idleThreads.insert_first "

"failed.");

}

PEGASUS_NAMESPACE_END

Legend:

Removed from v.1.80
changed lines
	Added in v.1.81

No CVS admin address has been configured