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.77 and 1.90.2.9

version 1.77, 2005/03/10 00:38:17

version 1.90.2.9, 2006/07/29 15:54:07

Line 1

//%2005////////////////////////////////////////////////////////////////////////

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

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

Line 8

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

Line 36

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

#include <Pegasus/Common/IPC.h>

#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

//==============================================================================

// POSIX Threads Implementation:

//==============================================================================

void thread_data::default_delete(void * data)

#if defined(PEGASUS_HAVE_PTHREADS)

{

if( data != NULL)

::operator delete(data);

}

// l10n start

struct StartWrapperArg

void language_delete(void * data)

{

if( data != NULL)

void *(PEGASUS_THREAD_CDECL * start) (void *);

{

void *arg;

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

};

}

// l10n end

Boolean Thread::_signals_blocked = false;

extern "C" void *_start_wrapper(void *arg_)

// l10n

{

#ifndef PEGASUS_OS_ZOS

StartWrapperArg *arg = (StartWrapperArg *) arg_;

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

#else

PEGASUS_THREAD_KEY_TYPE Thread::_platform_thread_key;

#endif

Boolean Thread::_key_initialized = false;

Boolean Thread::_key_error = false;

void *return_value = (*arg->start) (arg->arg);

delete arg;

// for non-native implementations

return return_value;

#ifndef PEGASUS_THREAD_CLEANUP_NATIVE

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

{

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

_cleanup.insert_first(cu.get());

cu.release();

return;

}

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

void Thread::cancel()

{

AutoPtr<cleanup_handler> cu ;

try

{

cu.reset(_cleanup.remove_first());

}

catch(IPCException&)

{

PEGASUS_ASSERT(0);

_cancelled = true;

pthread_cancel(_handle.thid.tt_handle());

}

if(execute == true)

cu->execute();

}

#endif

void Thread::test_cancel()

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

#ifndef PEGASUS_THREAD_EXIT_NATIVE

void Thread::exit_self(PEGASUS_THREAD_RETURN exit_code)

{

// execute the cleanup stack and then return

while( _cleanup.count() )

{

try

{

cleanup_pop(true);

#if defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)

}

pthread_testintr();

catch(IPCException&)

#else

{

pthread_testcancel();

PEGASUS_ASSERT(0);

break;

}

_exit_code = exit_code;

exit_thread(exit_code);

_handle.thid = 0;

}

#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 (pegasus_key_create(&Thread::_platform_thread_key) == 0)

Boolean Thread::is_cancelled(void)

{

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

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

Thread::_key_initialized = true;

}

else

{

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

return _cancelled;

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

Thread::_key_error = true;

return -1;

}

PEG_METHOD_EXIT();

void Thread::thread_switch()

return 0;

}

Thread * Thread::getCurrent()

{

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

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

{

return NULL;

#if defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)

}

pthread_yield(NULL);

PEG_METHOD_EXIT();

#else

return (Thread *)pegasus_get_thread_specific(_platform_thread_key);

sched_yield();

#endif

}

void Thread::setCurrent(Thread * thrd)

{

ATTN: why are these missing on other platforms?

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

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

#if defined(PEGASUS_PLATFORM_LINUX_GENERIC_GNU)

{

void Thread::suspend()

if (pegasus_set_thread_specific(Thread::_platform_thread_key,

(void *) thrd) == 0)

{

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

pthread_kill(_handle.thid.tt_handle(), SIGSTOP);

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

}

else

void Thread::resume()

{

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

pthread_kill(_handle.thid.tt_handle(), SIGCONT);

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

}

PEG_METHOD_EXIT();

}

#endif

AcceptLanguages * Thread::getLanguages()

void Thread::sleep(Uint32 msec)

{

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

Threads::sleep(msec);

Thread * curThrd = Thread::getCurrent();

if (curThrd == NULL)

return NULL;

AcceptLanguages * acceptLangs =

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

curThrd->dereference_tsd();

PEG_METHOD_EXIT();

return acceptLangs;

}

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

void Thread::join(void)

{

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

if (!_is_detached && Threads::id(_handle.thid) != 0)

pthread_join(_handle.thid.tt_handle(), &_exit_code);

Thread * currentThrd = Thread::getCurrent();

Threads::clear(_handle.thid);

if (currentThrd != NULL)

{

// deletes the old tsd and creates a new one

currentThrd->put_tsd("acceptLanguages",

language_delete,

sizeof(AcceptLanguages *),

langs);

}

PEG_METHOD_EXIT();

void Thread::thread_init(void)

{

#if defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)

pthread_setintr(PTHREAD_INTR_ENABLE);

pthread_setintrtype(PTHREAD_INTR_ASYNCHRONOUS);

#else

pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);

pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);

#endif

_cancel_enabled = true;

}

void Thread::clearLanguages() //l10n

void Thread::detach(void)

{

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

_is_detached = true;

pthread_detach(_handle.thid.tt_handle());

}

Thread * currentThrd = Thread::getCurrent();

ThreadStatus Thread::run()

if (currentThrd != NULL)

{

// deletes the old tsd

StartWrapperArg *arg = new StartWrapperArg;

currentThrd->delete_tsd("acceptLanguages");

arg->start = _start;

}

arg->arg = this;

PEG_METHOD_EXIT();

Threads::Type type = _is_detached ? Threads::DETACHED : Threads::JOINABLE;

}

int rc = Threads::create(_handle.thid, type, _start_wrapper, arg);

// l10n end

#if 0

// On Linux distributions released prior 2005, the implementation of

// two special synchronization classes for ThreadPool

// Native POSIX Thread Library returns ENOMEM instead of EAGAIN when

// there

// are no insufficient memory. Hence we are checking for both. See bug

// 386.

class timed_mutex

if ((rc == EAGAIN) || (rc == ENOMEM))

{

public:

timed_mutex(Mutex* mut, int msec)

:_mut(mut)

{

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

}

~timed_mutex(void)

{

_mut->unlock();

Threads::clear(_handle.thid);

delete arg;

return PEGASUS_THREAD_INSUFFICIENT_RESOURCES;

}

Mutex* _mut;

else if (rc != 0)

};

#endif

class try_mutex

{

public:

try_mutex(Mutex* mut)

:_mut(mut)

{

_mut->try_lock(pegasus_thread_self());

Threads::clear(_handle.thid);

delete arg;

return PEGASUS_THREAD_SETUP_FAILURE;

}

~try_mutex(void)

return PEGASUS_THREAD_OK;

{

_mut->unlock();

}

Mutex* _mut;

static sigset_t *block_signal_mask(sigset_t * sig)

};

class auto_int

{

public:

sigemptyset(sig);

auto_int(AtomicInt* num)

// should not be used for main()

: _int(num)

sigaddset(sig, SIGHUP);

{

sigaddset(sig, SIGINT);

_int->operator++();

// maybe useless, since KILL can't be blocked according to POSIX

}

sigaddset(sig, SIGKILL);

~auto_int(void)

{

_int->operator--();

}

AtomicInt *_int;

};

sigaddset(sig, SIGABRT);

sigaddset(sig, SIGALRM);

sigaddset(sig, SIGPIPE);

AtomicInt _idle_control;

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

// Note: older versions of the linux pthreads library use SIGUSR1 and SIGUSR2

// internally to stop and start threads that are blocking, the newer ones

// implement this through the kernel's real time signals

// since SIGSTOP/CONT can handle suspend()/resume() on Linux

// block them

// #if defined(PEGASUS_PLATFORM_LINUX_IX86_GNU)

// sigaddset(sig, SIGUSR1);

// sigaddset(sig, SIGUSR2);

// #endif

#ifndef PEGASUS_PLATFORM_ZOS_ZSERIES_IBM

pthread_sigmask(SIG_BLOCK, sig, NULL);

#else

sigprocmask(SIG_BLOCK, sig, NULL);

#endif

return sig;

}

void ThreadPool::kill_idle_threads(void)

Thread::Thread(ThreadReturnType(PEGASUS_THREAD_CDECL * start) (void *), void *parameter, Boolean detached):_is_detached(detached),

_cancel_enabled(true),

_cancelled(false),

_start(start), _cleanup(), _tsd(), _thread_parm(parameter), _exit_code(0)

{

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

Threads::clear(_handle.thid);

}

pegasus_gettimeofday(&now);

Thread::~Thread()

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

{

_pools.lock();

ThreadPool *p = _pools.next(0);

while(p != 0)

{

try

{

p->kill_dead_threads();

join();

empty_tsd();

}

catch(...)

{

// Do not allow the destructor to throw an exception

}

p = _pools.next(p);

}

_pools.unlock();

pegasus_gettimeofday(&last);

}

ThreadPool::ThreadPool(Sint16 initial_size,

#endif /* PEGASUS_HAVE_PTHREADS */

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

_dead(true), _dying(0)

{

_allocate_wait.tv_sec = alloc_wait.tv_sec;

_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(i = 0; i < initial_size; i++)

{

_link_pool(_init_thread());

}

_pools.insert_last(this);

}

//==============================================================================

// Windows Threads Implementation:

//==============================================================================

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

#if defined(PEGASUS_HAVE_WINDOWS_THREADS)

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

ThreadStatus Thread::run(void)

{

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

try

{

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

// Note: A Win32 thread ID is not the same thing as a pthread ID.

_dying++;

// Win32 threads have both a thread ID and a handle. The handle

// is used in the wait functions, etc.

// So _handle.thid is actually the thread handle.

// remove from the global pools list

unsigned threadid = 0;

_pools.remove(this);

// start with idle threads.

ThreadType tt;

Thread *th = 0;

tt.handle = (HANDLE) _beginthreadex(NULL, 0, _start, this, 0, &threadid);

th = _pool.remove_first();

_handle.thid = tt;

Semaphore* sleep_sem;

while(th != 0)

if (Threads::id(_handle.thid) == 0)

{

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

if (errno == EAGAIN)

PEGASUS_ASSERT(sleep_sem != 0);

if(sleep_sem == 0)

{

th->dereference_tsd();

return PEGASUS_THREAD_INSUFFICIENT_RESOURCES;

}

else

{

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

return PEGASUS_THREAD_SETUP_FAILURE;

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

}

return PEGASUS_THREAD_OK;

}

while(_idle_control.value())

void Thread::cancel(void)

pegasus_yield();

th = _dead.remove_first();

while(th != 0)

{

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

_cancelled = true;

PEGASUS_ASSERT(sleep_sem != 0);

}

if(sleep_sem == 0)

void Thread::test_cancel(void)

{

th->dereference_tsd();

if (_cancel_enabled && _cancelled)

}

else

{

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

exit_self(0);

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

th->join();

delete th;

}

th = _dead.remove_first();

}

Boolean Thread::is_cancelled(void)

{

th = _running.remove_first();

return _cancelled;

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

void Thread::thread_switch(void)

delete th;

{

}

Sleep(0);

th = _running.remove_first();

}

catch(...)

void Thread::sleep(Uint32 milliseconds)

{

}

Sleep(milliseconds);

}

// make this static to the class

void Thread::join(void)

PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL ThreadPool::_loop(void *parm)

{

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

if (Threads::id(_handle.thid) != 0)

Thread *myself = (Thread *)parm;

if(myself == 0)

{

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

if (!_is_detached)

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

{

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

if (!_cancelled)

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

PEG_METHOD_EXIT();

throw NullPointer();

}

if(pool->_dying.value())

{

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

// Emulate the unix join api. Caller sleeps until thread is

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

// done.

PEG_METHOD_EXIT();

WaitForSingleObject(_handle.thid.handle, INFINITE);

return((PEGASUS_THREAD_RETURN)0);

}

else

Semaphore *sleep_sem = 0;

Semaphore *blocking_sem = 0;

struct timeval *deadlock_timer = 0;

try

{

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

// Currently this is the only way to ensure this code does

myself->dereference_tsd();

// not

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

// hang forever.

myself->dereference_tsd();

if (WaitForSingleObject(_handle.thid.handle, 10000) ==

}

WAIT_TIMEOUT)

catch(...)

{

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

TerminateThread(_handle.thid.handle, 0);

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

_graveyard(myself);

PEG_METHOD_EXIT();

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

return((PEGASUS_THREAD_RETURN)0);

}

while(1)

DWORD exit_code = 0;

{

GetExitCodeThread(_handle.thid.handle, &exit_code);

if(pool->_dying.value())

_exit_code = (ThreadReturnType) exit_code;

break;

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.

CloseHandle(_handle.thid.handle);

Threads::clear(_handle.thid);

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

void Thread::thread_init(void)

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

try

{

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

_cancel_enabled = true;

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

{

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

}

if(_work == 0)

void Thread::detach(void)

{

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

_is_detached = true;

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

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

}

if(_work ==

Thread::Thread(ThreadReturnType(PEGASUS_THREAD_CDECL * start) (void *),

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

void *parameter,

Boolean detached):_is_detached(detached),

_cancel_enabled(true),

_cancelled(false),

_start(start), _cleanup(), _tsd(), _thread_parm(parameter), _exit_code(0)

{

Threads::clear(_handle.thid);

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

Thread::~Thread()

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

{

try

{

PEG_TRACE_STRING(TRC_THREAD, Tracer::LEVEL4,

join();

"Worker started");

empty_tsd();

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

}

#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

#endif /* PEGASUS_HAVE_WINDOWS_THREADS */

try

{

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

{

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 )

// Common implementation:

{

pool->_pool.insert_first(myself);

//==============================================================================

}

else

void thread_data::default_delete(void *data)

{

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

if (data != NULL)

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

::operator delete(data);

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

}

else

void language_delete(void *data)

{

PEG_METHOD_EXIT();

if (data != NULL)

return((PEGASUS_THREAD_RETURN)0);

}

catch(...)

{

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

AutoPtr < AcceptLanguageList > al(static_cast <

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

AcceptLanguageList * >(data));

PEG_METHOD_EXIT();

return((PEGASUS_THREAD_RETURN)0);

}

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

Boolean Thread::_signals_blocked = false;

// wait to be awakend by the thread pool destructor

#ifndef PEGASUS_OS_ZOS

//sleep_sem->wait();

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

#else

myself->test_cancel();

TSDKeyType Thread::_platform_thread_key;

#endif

PEG_METHOD_EXIT();

Boolean Thread::_key_initialized = false;

return((PEGASUS_THREAD_RETURN)0);

Boolean Thread::_key_error = false;

}

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

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

try

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

{

if (_dying.value())

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

{

_cleanup.insert_front(cu.get());

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

cu.release();

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

return;

// ATTN: Error result has not yet been defined

return true;

}

struct timeval start;

gettimeofday(&start, NULL);

Thread *th = 0;

th = _pool.remove_first();

if (th == 0)

void Thread::cleanup_pop(Boolean execute)

{

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

AutoPtr < cleanup_handler > cu;

{

try

th = _init_thread();

}

if (th == 0)

{

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

cu.reset(_cleanup.remove_front());

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

}

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

parm);

th->delete_tsd("work func");

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

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

(void *)work);

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

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

}

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

catch(IPCException &)

sleep_sem->signal();

th->dereference_tsd();

}

catch (...)

{

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

PEGASUS_ASSERT(0);

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

PEG_METHOD_EXIT();

// ATTN: Error result has not yet been defined

return true;

}

PEG_METHOD_EXIT();

if (execute == true)

return true;

cu->execute();

}

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

Uint32 ThreadPool::kill_dead_threads(void)

throw(IPCException)

{

// Since the kill_dead_threads, ThreadPool or allocate_and_awaken

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

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

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

try

void Thread::exit_self(ThreadReturnType exit_code)

{

if (_dying.value())

#if defined(PEGASUS_PLATFORM_HPUX_ACC) || \

defined(PEGASUS_PLATFORM_LINUX_GENERIC_GNU)

// NOTE: pthread_exit exhibits unusual behavior on RHEL 3 U2, as

// documented in Bugzilla 3836. Where feasible, it may be advantageous

// to avoid using this function.

pthread_exit(exit_code);

#else

// execute the cleanup stack and then return

while (_cleanup.size())

{

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

cleanup_pop(true);

{

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

Thread *dead = _dead.remove_first();

if(dead )

{

dead->join();

delete dead;

}

catch(...)

{

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

}

catch(IPCException &)

if (_dying.value())

{

return 0;

PEGASUS_ASSERT(0);

break;

}

Thread *th = 0;

internal_dq idq;

if(_pool.count() > 0 )

{

try

{

_pool.try_lock();

}

catch(...)

_exit_code = exit_code;

{

Threads::exit(exit_code);

return bodies;

Threads::clear(_handle.thid);

#endif

}

struct timeval dt = { 0, 0 };

Sint8 Thread::initializeKey()

struct timeval *dtp;

th = _pool.next(th);

while (th != 0 )

{

try

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

if (!Thread::_key_initialized)

{

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

if (Thread::_key_error)

}

catch(...)

{

_pool.unlock();

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

return bodies;

"Thread: ERROR - thread key error");

return -1;

}

if(dtp != 0)

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

{

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

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

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

Thread::_key_initialized = true;

}

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

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

th = _pool.next(th);

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

continue;

Thread::_key_error = true;

}

return -1;

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

idq.insert_first((void*)th);

}

th = _pool.next(th);

}

_pool.unlock();

}

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

while(th != 0)

{

th->delete_tsd("work func");

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

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

(void *)&_undertaker);

th->delete_tsd("work parm");

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

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

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

PEGASUS_ASSERT(sleep_sem != 0);

bodies++;

th->dereference_tsd();

sleep_sem->signal();

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

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

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

}

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

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

while (needed.value() > 0) {

_link_pool(_init_thread());

needed--;

pegasus_sleep(0);

}

return bodies;

}

catch (...)

{

}

PEG_METHOD_EXIT();

return 0;

}

Thread *Thread::getCurrent()

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

{

// never time out if the interval is zero

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

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

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

return false;

struct timeval now , finish , remaining ;

Uint32 usec;

pegasus_gettimeofday(&now);

/* remove valgrind error */

pegasus_gettimeofday(&remaining);

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;

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

return true;

else

return false;

}

PEGASUS_THREAD_RETURN ThreadPool::_undertaker( void *parm )

{

return NULL;

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

}

exit_thread((PEGASUS_THREAD_RETURN)1);

PEG_METHOD_EXIT();

return (PEGASUS_THREAD_RETURN)1;

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

}

PEGASUS_THREAD_RETURN ThreadPool::_graveyard(Thread *t)

void Thread::setCurrent(Thread * thrd)

{

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

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

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

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

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)

if (TSDKey::

set_thread_specific(Thread::_platform_thread_key,

(void *) thrd) == 0)

{

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

"Moving thread %p", t);

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

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

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

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)

AcceptLanguageList *Thread::getLanguages()

{

if(p != 0)

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

{

delete (Semaphore *)p;

}

Boolean ThreadPool::_check_dealloc(struct timeval *start)

Thread *curThrd = Thread::getCurrent();

{

if (curThrd == NULL)

return(check_time(start, &_deallocate_wait));

return NULL;

AcceptLanguageList *acceptLangs =

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

curThrd->dereference_tsd();

PEG_METHOD_EXIT();

return acceptLangs;

}

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

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

{

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

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

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

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

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

if (!th->run())

Thread *currentThrd = Thread::getCurrent();

if (currentThrd != NULL)

{

delete th;

// deletes the old tsd and creates a new one

return 0;

currentThrd->put_tsd("acceptLanguages",

language_delete,

sizeof (AcceptLanguageList *), langs);

}

_current_threads++;

pegasus_yield();

PEG_METHOD_EXIT();

return th;

PEG_METHOD_EXIT();

}

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

void Thread::clearLanguages() // l10n

{

if(th == 0)

{

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

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

throw NullPointer();

}

try

{

_pool.insert_first(th);

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

}

catch(...)

Thread *currentThrd = Thread::getCurrent();

if (currentThrd != NULL)

{

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

// deletes the old tsd

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

currentThrd->delete_tsd("acceptLanguages");

}

PEG_METHOD_EXIT();

}

// ATTN: not sure where to put this!

#ifdef PEGASUS_ZOS_SECURITY

bool isEnhancedSecurity = 99;

#endif

PEGASUS_NAMESPACE_END

Legend:

Removed from v.1.77
changed lines
	Added in v.1.90.2.9

No CVS admin address has been configured