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

Return to Thread.h CVS log

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

Diff for /pegasus/src/Pegasus/Common/Thread.h between version 1.1.2.13 and 1.61

version 1.1.2.13, 2001/10/03 16:55:03

version 1.61, 2008/03/20 01:06:41

Line 1

//%///////////-*-c++-*-//////////////////////////////////////////////////////

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

Line 29

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

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

// Modified By: Markus Mueller

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

#ifndef Pegasus_Thread_h

#define Pegasus_Thread_h

#include <Pegasus/Common/IPC.h>

#include <cstring>

#include <Pegasus/Common/Config.h>

#include <Pegasus/Common/Exception.h>

#include <Pegasus/Common/AtomicInt.h>

#include <Pegasus/Common/DQueue.h>

#include <Pegasus/Common/InternalException.h>

#include <Pegasus/Common/AcceptLanguageList.h>

#include <Pegasus/Common/Linkage.h>

#include <Pegasus/Common/AutoPtr.h>

#include <Pegasus/Common/List.h>

#include <Pegasus/Common/Mutex.h>

#include <Pegasus/Common/Semaphore.h>

#include <Pegasus/Common/TSDKey.h>

#include <Pegasus/Common/Threads.h>

#if defined(PEGASUS_HAVE_PTHREADS)

# include <signal.h>

#endif

PEGASUS_NAMESPACE_BEGIN

class PEGASUS_EXPORT cleanup_handler

class PEGASUS_COMMON_LINKAGE cleanup_handler : public Linkable

{

public:

cleanup_handler( void (*routine)(void *), void *arg ) : _routine(routine), _arg(arg) {}

cleanup_handler(void (*routine) (void *), void *arg):_routine(routine),

~cleanup_handler() {; }

_arg(arg)

private:

void execute(void) { _routine(_arg); }

cleanup_handler();

void (*_routine)(void *);

inline Boolean operator==(const void *key) const

{

if(key == (void *)_routine)

return true;

return false;

}

inline Boolean operator ==(const cleanup_handler & b) const

~cleanup_handler()

{

{;

return(operator==((const void *)b._routine));

}

void *_arg;

PEGASUS_CLEANUP_HANDLE _cleanup_buffer;

friend class DQueue<class cleanup_handler>;

friend class Thread;

};

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

class PEGASUS_EXPORT SimpleThread

{

public:

SimpleThread( PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *start )(void *),

void *parameter, Boolean detached );

~SimpleThread();

void run(void);

Uint32 threadId(void);

// get the user parameter

void *get_parm(void);

// cancellation must be deferred (not asynchronous)

// for user-level threads the thread itself can decide

// when it should die.

void cancel(void);

void kill(int signum);

// cancel if there is a pending cancellation request

void test_cancel(void);

// for user-level threads - put the calling thread

// to sleep and jump to the thread scheduler.

// platforms with preemptive scheduling and native threads

// can define this to be a no-op.

// platforms without preemptive scheduling like NetWare

// or gnu portable threads will have an existing

// routine that can be mapped to this method

void thread_switch(void);

// suspend this thread

void suspend(void) ;

// resume this thread

void resume(void) ;

void sleep(Uint32 msec) ;

// block the calling thread until this thread terminates

void join( PEGASUS_THREAD_RETURN *ret_val);

// stack of functions to be called when thread terminates

// will be called last in first out (LIFO)

void cleanup_push( void (*routine) (void *), void *parm );

void cleanup_pop(Boolean execute) ;

PEGASUS_THREAD_TYPE self(void) ;

private:

SimpleThread();

PEGASUS_THREAD_HANDLE _handle;

Boolean _is_detached;

Boolean _cancel_enabled;

Boolean _cancelled;

//PEGASUS_SEM_HANDLE _suspend_count;

void execute()

Semaphore _suspend;

{

_routine(_arg);

}

// always pass this * as the void * parameter to the thread

cleanup_handler();

// store the user parameter in _thread_parm

PEGASUS_THREAD_RETURN ( PEGASUS_THREAD_CDECL *_start)(void *) ;

void (*_routine)(void*);

void *_arg;

void *_thread_parm;

friend class Thread;

} ;

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

class PEGASUS_COMMON_LINKAGE thread_data : public Linkable

class PEGASUS_EXPORT thread_data

{

public:

static void default_delete(void *data);

thread_data( Sint8 *key ) : _delete_func(NULL) , _data(NULL), _size(0)

thread_data(const char *key) : _delete_func(NULL), _data(NULL), _size(0)

{

PEGASUS_ASSERT(key != NULL);

size_t keysize = strlen(key);

_key = new Sint8 [keysize + 1];

_key.reset(new char[keysize + 1]);

memcpy(_key, key, keysize);

memcpy(_key.get(), key, keysize);

_key[keysize] = 0x00;

_key.get()[keysize] = 0x00;

}

thread_data(Sint8 *key, size_t size) : _delete_func(default_delete), _size(size)

thread_data(const char *key, size_t size) :

_delete_func(default_delete), _size(size)

{

PEGASUS_ASSERT(key != NULL);

size_t keysize = strlen(key);

_key = new Sint8 [keysize + 1];

_key.reset(new char[keysize + 1]);

memcpy(_key, key, keysize);

memcpy(_key.get(), key, keysize);

_key[keysize] = 0x00;

_key.get()[keysize] = 0x00;

_data = ::operator new(_size) ;

}

thread_data(Sint8 *key, size_t size, void *data) : _delete_func(default_delete), _size(size)

thread_data(const char *key, size_t size, void *data)

: _delete_func(default_delete), _size(size)

{

PEGASUS_ASSERT(key != NULL);

PEGASUS_ASSERT(data != NULL);

size_t keysize = strlen(key);

_key = new Sint8[keysize + 1];

_key.reset(new char[keysize + 1]);

memcpy(_key, key, keysize);

memcpy(_key.get(), key, keysize);

_key[keysize] = 0x00;

_key.get()[keysize] = 0x00;

_data = ::operator new(_size);

memcpy(_data, data, size);

}

Line 187

Line 125

{

if( _data != NULL)

if(_delete_func != NULL)

{

_delete_func( _data );

if( _key != NULL )

}

delete [] _key;

}

void put_data(void (*del)(void *), size_t size, void *data ) throw(NullPointer)

/**

* This function is used to put data in thread space.

* Be aware that there is NOTHING in place to stop

* other users of the thread to remove this data.

* Or change the data.

* You, the developer has to make sure that there are

* no situations in which this can arise (ie, have a

* lock for the function which manipulates the TSD.

* @exception NullPointer

void put_data(void (*del) (void *), size_t size, void* data)

{

if(_data != NULL)

if(_delete_func != NULL)

Line 204

Line 155

return ;

}

size_t get_size(void) { return _size; }

size_t get_size()

{

return _size;

}

/**

This function is used to retrieve data from the

TSD, the thread specific data.

Be aware that there is NOTHING in place to stop

other users of the thread to change the data you

get from this function.

You, the developer has to make sure that there are

no situations in which this can arise (ie, have a

lock for the function which manipulates the TSD.

void get_data(void **data, size_t *size)

{

if(data == NULL || size == NULL)

Line 213

Line 179

*data = _data;

*size = _size;

return;

}

void copy_data(void **buf, size_t *size) throw(BufferTooSmall, NullPointer)

// @exception NullPointer

void copy_data(void** buf, size_t* size)

{

if((buf == NULL) || (size == NULL))

throw NullPointer() ;

*buf = ::operator new(_size);

*size = _size;

memcpy(*buf, _data, _size);

return;

}

inline Boolean operator==(const void *key) const

{

if ( ! strcmp(_key, (Sint8 *)key))

if (!strcmp(_key.get(), reinterpret_cast < const char *>(key)))

return(true);

return true;

return(false);

return false;

}

inline Boolean operator==(const thread_data& b) const

{

return(operator==((const void *)b._key));

return operator==(b._key.get());

}

static bool equal(const thread_data* node, const void* key)

{

return ((thread_data *) node)->operator==(key);

}

private:

Line 243

Line 213

thread_data();

void *_data;

size_t _size;

Sint8 *_key;

AutoArrayPtr<char> _key;

friend class DQueue<thread_data>;

friend class Thread;

};

enum ThreadStatus

{

PEGASUS_THREAD_OK = 1, /* No problems */

PEGASUS_THREAD_INSUFFICIENT_RESOURCES, /* Can't allocate a thread.

Not enough memory. Try

again later */

PEGASUS_THREAD_SETUP_FAILURE, /* Could not allocate into the thread

specific data storage. */

PEGASUS_THREAD_UNAVAILABLE /* Service is being destroyed and no new

threads can be provided. */

};

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

class PEGASUS_EXPORT Thread

class PEGASUS_COMMON_LINKAGE Thread : public Linkable

{

public:

Thread( PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *start )(void *),

void *parameter, Boolean detached );

Thread(

ThreadReturnType(PEGASUS_THREAD_CDECL* start) (void*),

void* parameter,

Boolean detached);

~Thread();

void run(void);

/** Start the thread.

@return PEGASUS_THREAD_OK if the thread is started successfully,

PEGASUS_THREAD_INSUFFICIENT_RESOURCES if the resources necessary

to start the thread are not currently available.

PEGASUS_THREAD_SETUP_FAILURE if the thread could not

be create properly - check the 'errno' value for specific operating

system return code.

ThreadStatus run();

// get the user parameter

inline void *get_parm(void) { return _thread_parm; }

inline void *get_parm()

{

// send the thread a signal -- may not be appropriate due to Windows

return _thread_parm;

// void kill(int signum);

}

// cancellation must be deferred (not asynchronous)

// for user-level threads the thread itself can decide

// when it should die.

void cancel(void);

void cancel();

// cancel if there is a pending cancellation request

void test_cancel(void);

// for user-level threads - put the calling thread

// to sleep and jump to the thread scheduler.

Line 284

Line 273

// or gnu portable threads will have an existing

// routine that can be mapped to this method

void thread_switch(void);

void thread_switch();

#ifdef PEGASUS_PLATFORM_LINUX_IX86_GNU

// suspend this thread

void suspend(void) ;

// resume this thread

void resume(void) ;

#endif

void sleep(Uint32 msec) ;

static void sleep(Uint32 msec);

// block the calling thread until this thread terminates

void join( void );

void join();

void thread_init(void);

// thread routine needs to call this function when

// it is ready to exit

void exit_self(PEGASUS_THREAD_RETURN return_code) ;

void exit_self(ThreadReturnType return_code);

// stack of functions to be called when thread terminates

// will be called last in first out (LIFO)

void cleanup_push( void (*routine) (void *), void *parm ) throw(IPCException);

// @exception IPCException

void cleanup_pop(Boolean execute = true) throw(IPCException);

void cleanup_push(void (*routine) (void *), void* parm);

// @exception IPCException

void cleanup_pop(Boolean execute = true);

// create and initialize a tsd

inline void create_tsd(Sint8 *key, int size, void *buffer) throw(IPCException)

// @exception IPCException

inline void create_tsd(const char* key, int size, void* buffer)

{

thread_data *tsd = new thread_data(key, size, buffer);

AutoPtr<thread_data> tsd(new thread_data(key, size, buffer));

try { _tsd.insert_first(tsd); }

_tsd.insert_front(tsd.get());

catch(IPCException& e) { e = e; delete tsd; throw; }

tsd.release();

}

// get the buffer associated with the key

// NOTE: this call leaves the tsd LOCKED !!!!

inline void *reference_tsd(Sint8 *key) throw(IPCException)

// @exception IPCException

inline void *reference_tsd(const char* key)

{

_tsd.lock();

thread_data *tsd = _tsd.reference((void *)key);

thread_data *tsd = _tsd.find(thread_data::equal, key);

if(tsd != NULL)

return( (void *)(tsd->_data) );

return (void *) (tsd->_data);

else

return(NULL);

return NULL;

}

// @exception IPCException

inline void *try_reference_tsd(const char *key)

{

_tsd.try_lock();

thread_data *tsd = _tsd.find(thread_data::equal, key);

if (tsd != NULL)

return (void *) (tsd->_data);

else

return NULL;

}

// release the lock held on the tsd

// NOTE: assumes a corresponding and prior call to reference_tsd() !!!

inline void dereference_tsd(void) throw(IPCException)

// @exception IPCException

inline void dereference_tsd()

{

_tsd.unlock();

}

// delete the tsd associated with the key

inline void delete_tsd(Sint8 *key) throw(IPCException)

// @exception IPCException

inline void delete_tsd(const char *key)

{

thread_data *tsd = _tsd.remove((void *)key);

AutoPtr < thread_data > tsd(_tsd.remove(thread_data::equal, key));

if(tsd != NULL)

delete tsd;

}

// create or re-initialize tsd associated with the key

// @exception IPCException

// if the tsd already exists, return the existing buffer

inline void empty_tsd()

thread_data *put_tsd(Sint8 *key, void (*delete_func)(void *), Uint32 size, void *value)

{

throw(IPCException)

_tsd.clear();

}

// create or re-initialize tsd associated with the key

// if the tsd already exists, delete the existing buffer

// @exception IPCException

void put_tsd(

const char* key,

void (*delete_func) (void *),

Uint32 size,

void* value)

{

PEGASUS_ASSERT(key != NULL);

PEGASUS_ASSERT(delete_func != NULL);

AutoPtr<thread_data> tsd;

thread_data *tsd ;

// This may throw an IPCException

tsd = _tsd.remove((void *)key); // may throw an IPC exception

tsd.reset(_tsd.remove(thread_data::equal, key));

thread_data *ntsd = new thread_data(key);

tsd.reset();

AutoPtr<thread_data> ntsd(new thread_data(key));

ntsd->put_data(delete_func, size, value);

try { _tsd.insert_first(ntsd); }

// This may throw an IPCException

catch(IPCException& e) { e = e; delete ntsd; throw; }

_tsd.insert_front(ntsd.get());

return(tsd);

ntsd.release();

}

inline PEGASUS_THREAD_RETURN get_exit(void) { return _exit_code; }

inline PEGASUS_THREAD_TYPE self(void) {return pegasus_thread_self(); }

PEGASUS_THREAD_HANDLE getThreadHandle() {return _handle;}

inline ThreadReturnType get_exit()

{

return _exit_code;

}

inline ThreadType self()

{

return Threads::self();

}

private:

ThreadHandle getThreadHandle()

Thread();

inline void create_tsd(Sint8 *key ) throw(IPCException)

{

thread_data *tsd = new thread_data(key);

return _handle;

try { _tsd.insert_first(tsd); }

catch(IPCException& e) { e = e; delete tsd; throw; }

}

PEGASUS_THREAD_HANDLE _handle;

Boolean _is_detached;

Boolean _cancel_enabled;

Boolean _cancelled;

PEGASUS_SEM_HANDLE _suspend_count;

void detach();

// always pass this * as the void * parameter to the thread

// store the user parameter in _thread_parm

// Gets the Thread object associated with the caller's thread.

// Note: this may return NULL if no Thread object is associated

// with the caller's thread.

static Thread *getCurrent();

PEGASUS_THREAD_RETURN ( PEGASUS_THREAD_CDECL *_start)(void *) ;

DQueue<class cleanup_handler> _cleanup;

// Sets the Thread object associated with the caller's thread.

DQueue<class thread_data> _tsd;

// Note: the Thread object must be placed on the heap.

static void setCurrent(Thread* thrd);

void *_thread_parm;

PEGASUS_THREAD_RETURN _exit_code;

// Gets the AcceptLanguageList object associated with the caller's

static Boolean _signals_blocked;

// Thread.

} ;

// Note: this may return NULL if no Thread object, or no

// AcceptLanguageList object, is associated with the caller's thread.

static AcceptLanguageList* getLanguages();

// Sets the AcceptLanguageList object associated with the caller's

// Thread.

// Note: a Thread object must have been previously associated with

// the caller's thread.

static void setLanguages(const AcceptLanguageList& langs);

#if 0

class PEGASUS_EXPORT Aggregator {

// Removes the AcceptLanguageList object associated with the caller's

// Thread.

static void clearLanguages();

public:

private:

Thread();

Aggregator();

static Sint8 initializeKey();

~Aggregator();

void started(void);

// @exception IPCException

void completed(void);

inline void create_tsd(const char *key)

void remaining(int operations);

{

void put_result(CIMReference *ref);

AutoPtr<thread_data> tsd(new thread_data(key));

_tsd.insert_front(tsd.get());

tsd.release();

}

private:

ThreadHandle _handle;

int _reference_count;

Boolean _is_detached;

Boolean _cancelled;

// always pass this * as the void * parameter to the thread

// store the user parameter in _thread_parm

// keep track of the thread running this operation so we can kill

ThreadReturnType(PEGASUS_THREAD_CDECL* _start) (void *);

// it if necessary

List<cleanup_handler, Mutex> _cleanup;

Thread _owner;

List<thread_data, Mutex> _tsd;

// this is a phased aggregate. when it is complete is will

void* _thread_parm;

// be streamed to the client regardless of the state of

ThreadReturnType _exit_code;

// siblings

static Boolean _signals_blocked;

Boolean _is_phased;

static TSDKeyType _platform_thread_key;

static Boolean _key_initialized;

int _total_values;

static Boolean _key_error;

int _completed_values;

int _total_child_values;

int _completed_child_values;

int _completion_state;

struct timeval _last_update;

time_t lifetime;

Aggregator *_parent;

// children may be phased or not phased

DQueue _children;

// empty results that are filled by provider

DQueue _results;

// array of predicates for events and

// stored queries (cursors)

Array _filter;

} ;

#endif

PEGASUS_NAMESPACE_END

Legend:

Removed from v.1.1.2.13
changed lines
	Added in v.1.61

No CVS admin address has been configured