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.51 and 1.72

version 1.51, 2003/10/16 19:21:28

version 1.72, 2004/10/25 18:26:02

Line 1

//%/////////////////////////////////////////////////////////////////////////////

//%2004////////////////////////////////////////////////////////////////////////

// The Open Group, Tivoli Systems

// Company, L.P.; IBM Corp.; The Open Group; Tivoli Systems.

// IBM Corp.; EMC Corporation, The Open Group.

// IBM Corp.; EMC Corporation; VERITAS Software Corporation; The Open Group.

// Permission is hereby granted, free of charge, to any person obtaining a copy

// of this software and associated documentation files (the "Software"), to

Line 25

Line 29

// Modified By: Rudy Schuet (rudy.schuet@compaq.com) 11/12/01

// added nsk platform support

// Roger Kumpf, Hewlett-Packard Company (roger_kumpf@hp.com)

// Amit K Arora, IBM (amita@in.ibm.com) for PEP#101

//%/////////////////////////////////////////////////////////////////////////////

#include "Thread.h"

#include <exception>

#include <Pegasus/Common/IPC.h>

#include <Pegasus/Common/Tracer.h>

Line 42

Line 49

# error "Unsupported platform"

#endif

PEGASUS_USING_STD;

PEGASUS_NAMESPACE_BEGIN

Line 56

Line 64

{

if( data != NULL)

{

AcceptLanguages * al = static_cast<AcceptLanguages *>(data);

AutoPtr<AcceptLanguages> al(static_cast<AcceptLanguages *>(data));

delete al;

}

// l10n end

Boolean Thread::_signals_blocked = false;

// l10n

#ifndef PEGASUS_OS_ZOS

PEGASUS_THREAD_KEY_TYPE Thread::_platform_thread_key = -1;

#else

PEGASUS_THREAD_KEY_TYPE Thread::_platform_thread_key;

#endif

Boolean Thread::_key_initialized = false;

Boolean Thread::_key_error = false;

Line 73

Line 84

#ifndef PEGASUS_THREAD_CLEANUP_NATIVE

void Thread::cleanup_push( void (*routine)(void *), void *parm) throw(IPCException)

{

cleanup_handler *cu = new cleanup_handler(routine, parm);

AutoPtr<cleanup_handler> cu(new cleanup_handler(routine, parm));

try

_cleanup.insert_first(cu.get());

{

cu.release();

_cleanup.insert_first(cu);

}

catch(IPCException&)

{

delete cu;

throw;

}

return;

}

void Thread::cleanup_pop(Boolean execute) throw(IPCException)

{

cleanup_handler *cu ;

AutoPtr<cleanup_handler> cu ;

try

{

cu = _cleanup.remove_first() ;

cu.reset(_cleanup.remove_first());

}

catch(IPCException&)

{

Line 99

Line 103

}

if(execute == true)

cu->execute();

delete cu;

}

#endif

Line 241

Line 244

}

// l10n end

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

#if 0

// two special synchronization classes for ThreadPool

class timed_mutex

{

public:

timed_mutex(Mutex* mut, int msec)

:_mut(mut)

{

_mut->timed_lock(msec, pegasus_thread_self());

}

~timed_mutex(void)

{

_mut->unlock();

}

Mutex* _mut;

};

#endif

class try_mutex

{

public:

try_mutex(Mutex* mut)

:_mut(mut)

{

_mut->try_lock(pegasus_thread_self());

}

~try_mutex(void)

{

_mut->unlock();

}

Mutex* _mut;

};

class auto_int

{

public:

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)

{

Line 305

Line 361

}

// 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");

try

{

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

auto_mutex(&(this->_monitor));

_dying++;

}

// remove from the global pools list

_pools.remove(this);

// start with idle threads.

Thread *th = 0;

th = _pool.remove_first();

Semaphore* sleep_sem;

while(th != 0)

{

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

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

PEGASUS_ASSERT(sleep_sem != 0);

if(sleep_sem == 0)

{

th->dereference_tsd();

throw NullPointer();

}

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();

// signal the thread's sleep semaphore

th->cancel();

th->join();

th->empty_tsd();

delete th;

}

th = _pool.remove_first();

}

while(_idle_control.value())

pegasus_yield();

th = _dead.remove_first();

while(th != 0)

{

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

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

PEGASUS_ASSERT(sleep_sem != 0);

if(sleep_sem == 0)

{

th->dereference_tsd();

throw NullPointer();

}

else

{

//ATTN-DME-P3-20030322: _dead queue processing in

//ThreadPool::~ThreadPool is inconsistent with the

//processing in kill_dead_threads. Is this correct?

// signal the thread's sleep semaphore

sleep_sem->signal();

th->dereference_tsd();

// signal the thread's sleep semaphore

th->cancel();

th->join();

th->empty_tsd();

delete th;

}

th = _dead.remove_first();

}

{

auto_mutex(&(this->_monitor));

{

th = _running.remove_first();

while(th != 0)

{

// signal the thread's sleep semaphore

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

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

PEGASUS_ASSERT(sleep_sem != 0);

if(sleep_sem == 0 )

{

th->dereference_tsd();

throw NullPointer();

}

else

{

sleep_sem->signal();

th->dereference_tsd();

//th->cancel();

pegasus_yield();

th->cancel();

// ensure that th->run() has a chance to execute so that the join will not

// block

th->join();

th->empty_tsd();

delete th;

}

th = _running.remove_first();

}

catch(...)

{

}

// make this static to the class

Line 413

Line 480

Thread *myself = (Thread *)parm;

if(myself == 0)

{

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

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

PEG_METHOD_EXIT();

throw NullPointer();

}

Line 425

Line 494

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

if(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;

Line 441

Line 519

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

myself->dereference_tsd();

}

catch(IPCException &)

{

PEG_METHOD_EXIT();

return(0);

}

catch(...)

{

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

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

_graveyard(myself);

PEG_METHOD_EXIT();

return(0);

return((PEGASUS_THREAD_RETURN)0);

}

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();

throw NullPointer();

return((PEGASUS_THREAD_RETURN)0);

}

while(pool->_dying.value() < 1)

while(1)

{

sleep_sem->wait();

if(pool->_dying.value())

break;

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

try

{

Boolean ignoreInterrupt = false;

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& )

{

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

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

_graveyard(myself);

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

}

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

/* Hence no need to move the thread to the _dead queue, as the _running

* queue is only dused by kill_dead_threads which makes sure that the

* the threads are cleaned up (unlocking any locked lists in the TSD, etc)

* before killing it.

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

void *parm = 0;

Line 481

Line 591

}

catch(IPCException &)

{

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

"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

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

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

_graveyard( myself);

PEG_METHOD_EXIT();

return(0);

return((PEGASUS_THREAD_RETURN)0);

}

if(_work == 0)

{

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

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

PEG_METHOD_EXIT();

throw NullPointer();

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);

try

{

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

{

auto_mutex(&(pool->_monitor));

try

if(pool->_dying.value())

{

break;

PEG_TRACE_STRING(TRC_THREAD, Tracer::LEVEL4,

"Worker started");

_work(parm);

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: ") +

e.getMessage());

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

}

_work(parm);

#if !defined(PEGASUS_OS_LSB)

catch (exception& e)

{

PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,

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,

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

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

}

// put myself back onto the available list

try

{

auto_mutex(&(pool->_monitor));

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

{

gettimeofday(deadlock_timer, NULL);

if( blocking_sem != 0 )

blocking_sem->signal();

pool->_running.remove((void *)myself);

// 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);

}

catch(IPCException &)

else

{

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

}

catch(...)

{

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

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

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

}

Line 554

Line 708

myself->test_cancel();

PEG_METHOD_EXIT();

myself->exit_self(0);

return((PEGASUS_THREAD_RETURN)0);

}

void ThreadPool::allocate_and_awaken(void *parm,

Boolean ThreadPool::allocate_and_awaken(void *parm,

PEGASUS_THREAD_RETURN \

(PEGASUS_THREAD_CDECL *work)(void *),

Semaphore *blocking)

throw(IPCException)

{

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

struct timeval start;

gettimeofday(&start, NULL);

// Allocate_and_awaken will not run if the _dying flag is set.

Thread *th = 0;

// Once the lock is acquired, ~ThreadPool will not change

// the value of _dying until the lock is released.

try

{

auto_mutex(&(this->_monitor));

if(_dying.value())

{

return;

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

}

"ThreadPool::allocate_and_awaken: ThreadPool is dying(1).");

th = _pool.remove_first();

// ATTN: Error result has not yet been defined

}

return true;

catch(...)

{

return;

}

struct timeval start;

gettimeofday(&start, NULL);

Thread *th = 0;

th = _pool.remove_first();

// wait for the right interval and try again

if (th == 0)

while (th == 0 && _dying.value() < 1)

{

// will throw an IPCException&

_check_deadlock(&start) ;

Line 595

Line 746

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

{

th = _init_thread();

continue;

}

pegasus_yield();

try

{

auto_mutex(&(this->_monitor));

if(_dying.value())

{

return;

}

th = _pool.remove_first();

}

if (th == 0)

catch(...)

{

return ;

// ATTN-DME-P3-20031103: This trace message should not be

}

// be labeled TRC_DISCARDED_DATA, because it does not

// necessarily imply that a failure has occurred. However,

// this label is being used temporarily to help isolate

// the cause of client timeout problems.

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

"ThreadPool::allocate_and_awaken: Insufficient resources: "

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

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

return false;

}

if(_dying.value() < 1)

{

// initialize the thread data with the work function and parameters

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

"Initializing thread with work function and parameters: parm = %p",

Line 629

Line 778

th->delete_tsd("blocking sem");

if(blocking != 0 )

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

try

{

auto_mutex(&(this->_monitor));

if(_dying.value())

{

th->cancel();

th->join();

delete th;

return;

}

// put the thread on the running list

_running.insert_first(th);

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

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

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();

}

Line 659

Line 799

}

catch(...)

{

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

"ThreadPool::allocate_and_awaken: Operation Failed.");

PEG_METHOD_EXIT();

return;

// ATTN: Error result has not yet been defined

}

return true;

}

else

{

th->cancel();

th->join();

delete th;

}

PEG_METHOD_EXIT();

return true;

}

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

Line 681

Line 816

Uint32 ThreadPool::kill_dead_threads(void)

throw(IPCException)

{

struct timeval now;

// Since the kill_dead_threads, ThreadPool or allocate_and_awaken

gettimeofday(&now, NULL);

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

Uint32 bodies = 0;

// be allowed to run at the same time.

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

// << 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);

// first go thread the dead q and clean it up as much as possible

try

{

auto_mutex(&(this->_monitor));

if(_dying.value() )

{

return 0;

}

while(_dead.count() > 0 && _dying.value() == 0 )

struct timeval now;

gettimeofday(&now, NULL);

Uint32 bodies = 0;

AtomicInt needed(0);

// first go thread the dead q and clean it up as much as possible

try

{

while(_dying.value() == 0 && _dead.count() > 0)

{

Tracer::trace(TRC_THREAD, Tracer::LEVEL4, "ThreadPool:: removing and joining dead thread");

Thread *dead = _dead.remove_first();

if(dead == 0)

if(dead )

throw NullPointer();

{

dead->join();

delete dead;

}

catch(...)

{

Tracer::trace(TRC_THREAD, Tracer::LEVEL4, "Exception when deleting dead");

}

if (_dying.value())

DQueue<Thread> * map[2] =

{

&_pool, &_running

return 0;

};

}

DQueue<Thread> *q = 0;

Thread *th = 0;

int i = 0;

internal_dq idq;

AtomicInt needed(0);

#ifdef PEGASUS_DISABLE_KILLING_HUNG_THREADS

if(_pool.count() > 0 )

// This change prevents the thread pool from killing "hung" threads.

// The definition of a "hung" thread is one that has been on the run queue

// for longer than the time interval set when the thread pool was created.

// Cancelling "hung" threads has proven to be problematic.

// With this change the thread pool will not cancel "hung" threads. This

// may prevent a crash depending upon the state of the "hung" thread. In

// the case that the thread is actually hung, this change causes the

// thread resources not to be reclaimed.

// Idle threads, those that have not executed a routine for a time

// interval, continue to be destroyed. This is normal and should not

// cause any problems.

for( ; i < 1; i++)

#else

for( ; i < 2; i++)

#endif

{

auto_mutex(&(this->_monitor));

q = map[i];

if(q->count() > 0 )

{

try

{

if(_dying.value())

_pool.try_lock();

{

return bodies;

}

q->try_lock();

}

catch(...)

{

Line 759

Line 879

struct timeval dt = { 0, 0 };

struct timeval *dtp;

Thread *th = 0;

th = q->next(th);

th = _pool.next(th);

while (th != 0 )

{

try

Line 769

Line 889

}

catch(...)

{

q->unlock();

_pool.unlock();

return bodies;

}

Line 780

Line 900

th->dereference_tsd();

struct timeval deadlock_timeout;

Boolean too_long;

if( i == 0)

{

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

}

else

{

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

}

if( true == too_long)

{

// if we are deallocating from the pool, escape if we are

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

// down to the minimum thread count

_current_threads--;

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

{

if( i == 0)

{

_current_threads++;

th = q->next(th);

th = _pool.next(th);

continue;

}

else

{

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

// we are killing a hung thread and we will drop below the

idq.insert_first((void*)th);

// minimum. create another thread to make up for the one

// we are about to kill

needed++;

}

th = _pool.next(th);

}

_pool.unlock();

}

th = q->remove_no_lock((void *)th);

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

while(th != 0)

if(th != 0)

{

if( i == 0 )

{

th->delete_tsd("work func");

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

Line 826

Line 933

// 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)

{

q->unlock();

th->dereference_tsd();

throw NullPointer();

}

bodies++;

th->dereference_tsd();

_dead.insert_first(th);

sleep_sem->signal();

th = 0;

th->join(); // Note: Clean up the thread here rather than

}

delete th; // leave it sitting unused on the _dead queue

else

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

{

// deadlocked threads

Tracer::trace(TRC_THREAD, Tracer::LEVEL4, "Killing a deadlocked thread");

th->cancel();

delete th;

}

th = q->next(th);

pegasus_sleep(1);

}

q->unlock();

}

if(_dying.value() )

return bodies;

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

"We need %u new threads", needed.value());

while (needed.value() > 0) {

_link_pool(_init_thread());

needed--;

Line 865

Line 952

}

return bodies;

}

catch (...)

{

}

PEG_METHOD_EXIT();

return 0;

}

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

Line 894

Line 987

PEGASUS_THREAD_RETURN ThreadPool::_undertaker( void *parm )

{

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;

}

if (!pool->_dead.exists(t))

{

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)

{

Line 927

Line 1068

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

Line 940

Line 1082

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

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

th->run();

if (!th->run())

{

delete th;

return 0;

}

_current_threads++;

pegasus_yield();

PEG_METHOD_EXIT();

return th;

}

Line 950

Line 1097

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

{

if(th == 0)

{

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

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

throw NullPointer();

}

try

{

auto_mutex(&(this->_monitor));

if(_dying.value())

{

th->cancel();

th->join();

delete th;

}

_pool.insert_first(th);

}

catch(...)

{

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

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

}

Legend:

Removed from v.1.51
changed lines
	Added in v.1.72

No CVS admin address has been configured