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.1.2.9 and 1.90.2.4

version 1.1.2.9, 2001/10/25 15:41:01

version 1.90.2.4, 2006/07/28 21:22:01

Line 1

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

//%2006////////////////////////////////////////////////////////////////////////

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

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

// EMC Corporation; Symantec 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 22

Line 31

// Author: Mike Day (mdday@us.ibm.com)

// Modified By:

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

// Sean Keenan, Hewlett-Packard Company (sean.keenan@hp.com)

// David Dillard, VERITAS Software Corp.

// (david.dillard@veritas.com)

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

#include "Thread.h"

#include <Pegasus/Common/IPC.h>

#include <exception>

#include <Pegasus/Common/Tracer.h>

#include "Time.h"

#if defined(PEGASUS_OS_TYPE_WINDOWS)

# include "ThreadWindows.cpp"

#elif defined(PEGASUS_OS_TYPE_UNIX)

# include "ThreadUnix.cpp"

#elif defined(PEGASUS_OS_TYPE_NSK)

# include "ThreadNsk.cpp"

#elif defined(PEGASUS_OS_VMS)

# include "ThreadVms.cpp"

#else

# error "Unsupported platform"

#endif

PEGASUS_USING_STD;

PEGASUS_NAMESPACE_BEGIN

void thread_data::default_delete(void * data)

{

if( data != NULL)

::operator delete(data);

}

// l10n start

Boolean Thread::_signals_blocked = false;

void language_delete(void * data)

// for non-native implementations

#ifndef PEGASUS_THREAD_CLEANUP_NATIVE

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

{

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

if( data != NULL)

try

{

_cleanup.insert_first(cu);

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

}

catch(IPCException& e)

{

delete cu;

throw;

}

// l10n end

Boolean Thread::_signals_blocked = false;

// l10n

#ifndef PEGASUS_OS_ZOS

TSDKeyType Thread::_platform_thread_key = TSDKeyType(-1);

#else

TSDKeyType Thread::_platform_thread_key;

#endif

Boolean Thread::_key_initialized = false;

Boolean Thread::_key_error = false;

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

{

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

_cleanup.insert_front(cu.get());

cu.release();

return;

}

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

void Thread::cleanup_pop(Boolean execute)

{

cleanup_handler *cu ;

AutoPtr<cleanup_handler> cu;

try

{

cu = _cleanup.remove_first() ;

cu.reset(_cleanup.remove_front());

}

catch(IPCException& e)

catch(IPCException&)

{

PEGASUS_ASSERT(0);

}

if(execute == true)

cu->execute();

delete cu;

}

#endif

//thread_data *Thread::put_tsd(Sint8 *key, void (*delete_func)(void *), Uint32 size, void *value) throw(IPCException)

//thread_data *Thread::put_tsd(const Sint8 *key, void (*delete_func)(void *), Uint32 size, void *value)

#ifndef PEGASUS_THREAD_EXIT_NATIVE

void Thread::exit_self(PEGASUS_THREAD_RETURN exit_code)

void Thread::exit_self(ThreadReturnType exit_code)

{

// execute the cleanup stack and then return

while( _cleanup.count() )

while( _cleanup.size() )

{

try

{

cleanup_pop(true);

}

catch(IPCException& e)

catch(IPCException&)

{

PEGASUS_ASSERT(0);

break;

}

_exit_code = exit_code;

exit_thread(exit_code);

Threads::exit(exit_code);

Threads::clear(_handle.thid);

}

#endif

// l10n start

Sint8 Thread::initializeKey()

{

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

if (!Thread::_key_initialized)

{

if (Thread::_key_error)

{

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

"Thread: ERROR - thread key error");

return -1;

}

if (TSDKey::create(&Thread::_platform_thread_key) == 0)

{

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

"Thread: able to create a thread key");

Thread::_key_initialized = true;

}

else

{

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

"Thread: ERROR - unable to create a thread key");

Thread::_key_error = true;

return -1;

}

PEG_METHOD_EXIT();

return 0;

}

Thread * Thread::getCurrent()

{

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

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

{

return NULL;

}

PEG_METHOD_EXIT();

return (Thread *)TSDKey::get_thread_specific(_platform_thread_key);

}

void Thread::setCurrent(Thread * thrd)

{

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

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

{

if (TSDKey::set_thread_specific(

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

{

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

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

}

else

{

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

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

}

PEG_METHOD_EXIT();

}

AcceptLanguageList * Thread::getLanguages()

{

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

Thread * curThrd = Thread::getCurrent();

if (curThrd == NULL)

return NULL;

AcceptLanguageList * acceptLangs =

(AcceptLanguageList *)curThrd->reference_tsd("acceptLanguages");

curThrd->dereference_tsd();

PEG_METHOD_EXIT();

return acceptLangs;

}

void Thread::setLanguages(AcceptLanguageList *langs) //l10n

{

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

ThreadPool::ThreadPool(Sint16 initial_size,

Thread* currentThrd = Thread::getCurrent();

Sint16 max,

if (currentThrd != NULL)

Sint16 min,

{

Sint8 *key)

// deletes the old tsd and creates a new one

: _max_threads(max), _min_threads(min),

currentThrd->put_tsd("acceptLanguages",

_current_threads(0), _waiters(initial_size),

language_delete,

_pool_sem(0), _pool(true), _running(true),

sizeof(AcceptLanguageList *),

langs);

}

PEG_METHOD_EXIT();

}

void Thread::clearLanguages() //l10n

{

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

Thread * currentThrd = Thread::getCurrent();

if (currentThrd != NULL)

{

// deletes the old tsd

currentThrd->delete_tsd("acceptLanguages");

}

PEG_METHOD_EXIT();

}

// l10n end

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

// ThreadPool

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

ThreadPool::ThreadPool(

Sint16 initialSize,

const char* key,

Sint16 minThreads,

Sint16 maxThreads,

struct timeval& deallocateWait)

: _maxThreads(maxThreads),

_minThreads(minThreads),

_currentThreads(0),

_idleThreads(),

_runningThreads(),

_dying(0)

{

_allocate_wait.tv_sec = 1;

_deallocateWait.tv_sec = deallocateWait.tv_sec;

_allocate_wait.tv_usec = 0;

_deallocateWait.tv_usec = deallocateWait.tv_usec;

_deallocate_wait.tv_sec = 30;

_deallocate_wait.tv_usec = 0;

_deadlock_detect.tv_sec = 60;

_deadlock_detect.tv_usec = 0;

memset(_key, 0x00, 17);

if(key != 0)

{

strncpy(_key, key, 16);

if(_max_threads < initial_size)

}

_max_threads = initial_size;

if(_min_threads > initial_size)

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

_min_threads = initial_size;

{

_maxThreads = initialSize;

}

if (_minThreads > initialSize)

{

_minThreads = initialSize;

}

int i;

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

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

{

_link_pool(_init_thread());

_addToIdleThreadsQueue(_initializeThread());

}

ThreadPool::~ThreadPool(void)

ThreadPool::~ThreadPool()

{

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

try

{

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

_dying++;

Thread *th = _pool.remove_first();

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

while(th != 0)

"Cleaning up %d idle threads. ", _currentThreads.get());

printf("***BEFORE\n");

while (_currentThreads.get() > 0)

{

Thread* thread = _idleThreads.remove_front();

if (thread != 0)

{

printf("***INSIDE1\n");

_cleanupThread(thread);

_currentThreads--;

}

else

{

printf("***INSIDE2\n");

Threads::yield();

}

printf("***AFTER\n");

}

catch (...)

{

// signal the thread's sleep semaphore

th->cancel();

th->join();

th->empty_tsd();

delete th;

th = _pool.remove_first();

}

// make this static to the class

ThreadReturnType PEGASUS_THREAD_CDECL ThreadPool::_loop(void* parm)

PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL ThreadPool::_loop(void *parm)

{

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

try

{

Thread *myself = (Thread *)parm;

if(myself == 0)

PEGASUS_ASSERT(myself != 0);

throw NullPointer();

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

throw NullPointer();

Semaphore *sleep_sem;

Semaphore* sleep_sem = 0;

struct timeval *deadlock_timer;

struct timeval* lastActivityTime = 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(IPCException & e)

catch (...)

{

myself->exit_self(0);

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

"ThreadPool::_loop: Failure getting sleep_sem or "

"lastActivityTime.");

PEGASUS_ASSERT(false);

pool->_idleThreads.remove(myself);

pool->_currentThreads--;

PEG_METHOD_EXIT();

return((ThreadReturnType)1);

}

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

throw NullPointer();

while(pool->_dying < 1)

while (1)

{

try

{

myself->test_cancel();

sleep_sem->wait();

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

}

myself->test_cancel();

catch (...)

gettimeofday(deadlock_timer, NULL);

{

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

PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *_work)(void *);

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

void *parm;

PEGASUS_ASSERT(false);

pool->_idleThreads.remove(myself);

pool->_currentThreads--;

PEG_METHOD_EXIT();

return((ThreadReturnType)1);

}

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

// _idleThreads queue.

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

void* parm = 0;

Semaphore* blocking_sem = 0;

try

{

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

work = (ThreadReturnType (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 & e)

catch (...)

{

myself->exit_self(0);

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

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

"or blocking sem.");

PEGASUS_ASSERT(false);

pool->_idleThreads.remove(myself);

pool->_currentThreads--;

PEG_METHOD_EXIT();

return((ThreadReturnType)1);

}

if(_work == 0)

if (work == 0)

throw NullPointer();

{

_work(parm);

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

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

break;

}

Time::gettimeofday(lastActivityTime);

try

{

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

work(parm);

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

}

catch (Exception & e)

{

PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,

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

e.getMessage());

}

#if !defined(PEGASUS_OS_LSB)

catch (const exception& e)

{

PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,

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

e.what());

}

#endif

catch (...)

{

PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,

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

}

// put myself back onto the available list

try

{

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

Time::gettimeofday(lastActivityTime);

pool->_link_pool(myself);

if (blocking_sem != 0)

{

blocking_sem->signal();

}

pool->_runningThreads.remove(myself);

pool->_idleThreads.insert_front(myself);

}

catch(IPCException & e)

catch (...)

{

myself->exit_self(0);

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

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

PEGASUS_ASSERT(false);

pool->_currentThreads--;

PEG_METHOD_EXIT();

return((ThreadReturnType)1);

}

myself->exit_self(0);

}

return((PEGASUS_THREAD_RETURN)0);

catch (const Exception& e)

{

PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"Caught exception: \"" + e.getMessage() + "\". Exiting _loop.");

}

catch (...)

{

PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"Caught unrecognized exception. Exiting _loop.");

}

PEG_METHOD_EXIT();

return((ThreadReturnType)0);

}

void ThreadPool::allocate_and_awaken(void *parm,

ThreadStatus ThreadPool::allocate_and_awaken(

PEGASUS_THREAD_RETURN \

void* parm,

(PEGASUS_THREAD_CDECL *work)(void *))

ThreadReturnType (PEGASUS_THREAD_CDECL* work)(void *),

throw(IPCException)

Semaphore* blocking)

{

struct timeval start;

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

gettimeofday(&start, NULL);

Thread *th = _pool.remove_first();

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

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

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

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

try

{

try // we couldn't get a free thread from the pool

{

// wait for the right interval and try again

if (_dying.get())

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

{

_check_deadlock(&start);

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

Uint32 interval = _allocate_wait.tv_sec * 1000;

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

if(_allocate_wait.tv_usec > 0)

return PEGASUS_THREAD_UNAVAILABLE;

interval += (_deallocate_wait.tv_usec / 1000);

// will throw a timeout if no thread comes free

_pool_sem.time_wait(interval);

th = _pool.remove_first();

}

struct timeval start;

catch(TimeOut & to)

Time::gettimeofday(&start);

Thread* th = 0;

th = _idleThreads.remove_front();

if (th == 0)

{

if(_current_threads < _max_threads)

if ((_maxThreads == 0) ||

(_currentThreads.get() < Uint32(_maxThreads)))

{

th = _init_thread();

th = _initializeThread();

break;

}

// will throw a Deadlock Exception before falling out of the loop

_check_deadlock(&start);

} // while th == null

if(_dying < 1)

if (th == 0)

{

// 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",

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

return PEGASUS_THREAD_INSUFFICIENT_RESOURCES;

}

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

th->remove_tsd("work func");

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

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

parm);

th->delete_tsd("work func");

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

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

sizeof( ThreadReturnType (PEGASUS_THREAD_CDECL *)(void *)),

(void *)work);

th->remove_tsd("work parm");

th->delete_tsd("work parm");

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

th->delete_tsd("blocking sem");

if (blocking != 0)

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

// put the thread on the running list

_running.insert_first(th);

_runningThreads.insert_front(th);

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

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

if(sleep_sem == 0)

PEGASUS_ASSERT(sleep_sem != 0);

throw NullPointer();

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

sleep_sem->signal();

th->dereference_tsd();

}

else

catch (...)

_pool.insert_first(th);

{

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

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

PEG_METHOD_EXIT();

// ATTN: Error result has not yet been defined

return PEGASUS_THREAD_SETUP_FAILURE;

}

PEG_METHOD_EXIT();

return PEGASUS_THREAD_OK;

}

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

// but should call it at least once per _deadlock_detect with the running q

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

// and at least once per _deallocate_wait for the pool q

void ThreadPool::_kill_dead_threads(DQueue<Thread> *q, Boolean (*check)(struct timeval *))

Uint32 ThreadPool::cleanupIdleThreads()

throw(IPCException)

{

struct timeval now;

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

gettimeofday(&now, NULL);

DQueue<Thread> dead ;

Uint32 numThreadsCleanedUp = 0;

if(q->count() > 0 )

Uint32 numIdleThreads = _idleThreads.size();

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

{

try

// Do not dip below the minimum thread count

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

{

q->try_lock();

break;

}

catch(AlreadyLocked & a)

Thread* thread = _idleThreads.remove_back();

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

if (thread == 0)

{

return;

break;

}

Thread *context = 0;

struct timeval* lastActivityTime;

struct timeval dt = { 0, 0 };

struct timeval *dtp;

Thread *th = q->next(context);

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(AlreadyLocked & a)

catch (...)

{

context = th;

PEGASUS_ASSERT(false);

th = q->next(context);

_idleThreads.insert_back(thread);

continue;

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

if( true == check(&dt))

{

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

if(th != 0)

{

dead.insert_first(th);

_idleThreads.insert_front(thread);

th = 0;

}

context = th;

th = q->next(context);

}

q->unlock();

}

if(dead.count())

PEG_METHOD_EXIT();

{

return numThreadsCleanedUp;

Thread *th = dead.remove_first();

while(th != 0)

{

th->cancel();

th->join();

delete th;

th = dead.remove_first();

}

return;

}

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

void ThreadPool::_cleanupThread(Thread* thread)

{

struct timeval now;

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

gettimeofday(&now, NULL);

if( (now.tv_sec - start->tv_sec) > interval->tv_sec ||

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

(((now.tv_sec - start->tv_sec) == interval->tv_sec) &&

thread->delete_tsd("work func");

((now.tv_usec - start->tv_usec) >= interval->tv_usec ) ) )

thread->put_tsd(

return true;

"work func", 0,

else

sizeof(ThreadReturnType (PEGASUS_THREAD_CDECL *)(void *)),

return false;

(void *) 0);

thread->delete_tsd("work parm");

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

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

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

PEGASUS_ASSERT(sleep_sem != 0);

sleep_sem->signal();

thread->dereference_tsd();

thread->join();

delete thread;

PEG_METHOD_EXIT();

}

void ThreadPool::_sleep_sem_del(void *p)

Boolean ThreadPool::_timeIntervalExpired(

struct timeval* start,

struct timeval* interval)

{

if(p != 0)

// never time out if the interval is zero

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

{

delete (Semaphore *)p;

return false;

}

inline void ThreadPool::_check_deadlock(struct timeval *start) throw(Deadlock)

struct timeval now, finish, remaining;

{

Uint32 usec;

if (true == _check_time(start, &_deadlock_detect))

Time::gettimeofday(&now);

throw Deadlock(pegasus_thread_self());

Time::gettimeofday(&remaining); // Avoid valgrind error

return;

}

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

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

finish.tv_sec += (usec / 1000000);

usec %= 1000000;

finish.tv_usec = usec;

inline Boolean ThreadPool::_check_deadlock_no_throw(struct timeval *start)

return (Time::subtract(&remaining, &finish, &now) != 0);

{

return(_check_time(start, &_deadlock_detect));

}

inline Boolean ThreadPool::_check_dealloc(struct timeval *start)

void ThreadPool::_deleteSemaphore(void *p)

{

return(_check_time(start, &_deallocate_wait));

delete (Semaphore *)p;

}

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

Thread* ThreadPool::_initializeThread()

{

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

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

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(

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

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

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

struct timeval* lastActivityTime =

th->run();

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

_current_threads++;

Time::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() != PEGASUS_THREAD_OK)

{

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

"Could not create thread. Error code is %d.", errno);

delete th;

return 0;

}

_currentThreads++;

Threads::yield();

PEG_METHOD_EXIT();

return th;

}

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

void ThreadPool::_addToIdleThreadsQueue(Thread* th)

{

if(th == 0)

{

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

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

throw NullPointer();

_pool.insert_first(th);

_pool_sem.signal();

}

try

{

_idleThreads.insert_front(th);

}

catch (...)

{

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

"ThreadPool::_addToIdleThreadsQueue: _idleThreads.insert_front "

"failed.");

}

// ATTN: not sure where to put this!

#ifdef PEGASUS_ZOS_SECURITY

bool isEnhancedSecurity=99;

#endif

PEGASUS_NAMESPACE_END

Legend:

Removed from v.1.1.2.9
changed lines
	Added in v.1.90.2.4

No CVS admin address has been configured