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

Return to MessageQueueService.cpp CVS log

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

Diff for /pegasus/src/Pegasus/Common/MessageQueueService.cpp between version 1.70.2.1 and 1.88

version 1.70.2.1, 2003/08/08 17:39:38

version 1.88, 2004/10/17 20:39:17

Line 1

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

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

Line 28

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

// Modified By:

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

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

#include "MessageQueueService.h"

#include <Pegasus/Common/Tracer.h>

#include <Pegasus/Common/MessageLoader.h> //l10n

PEGASUS_NAMESPACE_BEGIN

Line 53

Line 59

return _thread_pool;

}

void unload_idle_providers(void);

PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL MessageQueueService::kill_idle_threads(void *parm)

{

Line 74

Line 78

}

exit_thread((PEGASUS_THREAD_RETURN)dead_threads);

return (PEGASUS_THREAD_RETURN)dead_threads;

}

void MessageQueueService::force_shutdown(void)

void MessageQueueService::force_shutdown(Boolean destroy_flag)

{

PEGASUS_STD(cout) << "Forcing shutdown of CIMOM Message Router" << PEGASUS_STD(endl);

return;

MessageQueueService::_stop_polling = 1;

MessageQueueService *svc;

#ifdef MESSAGEQUEUESERVICE_DEBUG

//l10n

MessageLoaderParms parms("Common.MessageQueueService.FORCING_SHUTDOWN",

"Forcing shutdown of CIMOM Message Router");

PEGASUS_STD(cout) << MessageLoader::getMessage(parms) << PEGASUS_STD(endl);

#endif

MessageQueueService *svc;

int counter = 0;

_polling_list.lock();

svc = _polling_list.next(0);

while(svc != 0)

{

PEGASUS_STD(cout) << "Stopping " << svc->getQueueName() << PEGASUS_STD(endl);

#ifdef MESSAGEQUEUESERVICE_DEBUG

//l10n - reuse same MessageLoaderParms to avoid multiple creates

parms.msg_id = "Common.MessageQueueService.STOPPING_SERVICE";

parms.default_msg = "Stopping $0";

parms.arg0 = svc->getQueueName();

PEGASUS_STD(cout) << MessageLoader::getMessage(parms) << PEGASUS_STD(endl);

#endif

_polling_sem.signal();

svc->_shutdown_incoming_queue();

counter++;

_polling_sem.signal();

svc = _polling_list.next(svc);

}

_polling_list.unlock();

_polling_sem.signal();

MessageQueueService::_stop_polling = 1;

if(destroy_flag == true)

{

svc = _polling_list.remove_last();

while(svc)

{

delete svc;

svc = _polling_list.remove_last();

}

Line 125

Line 161

if(_check_idle_flag.value() != 0 )

{

_check_idle_flag = 0;

Thread th(kill_idle_threads, 0, true);

th.run();

// If there are insufficent resources to run

// kill_idle_threads, then just return.

_thread_pool->allocate_and_awaken(service, kill_idle_threads);

}

myself->exit_self( (PEGASUS_THREAD_RETURN) 1 );

Line 161

Line 199

_default_op_timeout.tv_sec = 30;

_default_op_timeout.tv_usec = 100;

_meta_dispatcher_mutex.lock(pegasus_thread_self());

AutoMutex autoMut(_meta_dispatcher_mutex);

if( _meta_dispatcher == 0 )

{

Line 169

Line 207

_meta_dispatcher = new cimom();

if (_meta_dispatcher == NULL )

{

_meta_dispatcher_mutex.unlock();

throw NullPointer();

}

_thread_pool = new ThreadPool(0, "MessageQueueService", 0, 0,

Line 178

Line 215

_polling_thread = new Thread(polling_routine,

reinterpret_cast<void *>(&_polling_list),

false);

_polling_thread->run();

while (!_polling_thread->run())

{

pegasus_yield();

}

_service_count++;

if( false == register_service(name, _capabilities, _mask) )

{

_meta_dispatcher_mutex.unlock();

//l10n

throw BindFailedException("MessageQueueService Base Unable to register with Meta Dispatcher");

//throw BindFailedException("MessageQueueService Base Unable to register with Meta Dispatcher");

MessageLoaderParms parms("Common.MessageQueueService.UNABLE_TO_REGISTER",

"MessageQueueService Base Unable to register with Meta Dispatcher");

throw BindFailedException(parms);

}

_polling_list.insert_last(this);

_meta_dispatcher_mutex.unlock();

// _meta_dispatcher_mutex.unlock(); //Bug#1090

// _callback_thread.run();

// _req_thread.run();

Line 200

Line 244

MessageQueueService::~MessageQueueService(void)

{

_die = 1;

// IBM-KR: This causes a new message (IO_CLOSE) to be spawned, which

// doesn't get picked up anyone. The idea was that the message would be

if (_incoming_queue_shutdown.value() == 0 )

// picked up handle_AsyncIoctl which closes the queue and does cleaning.

{

// That described behavior has never surfaced itself. If it does appear,

_shutdown_incoming_queue();

// uncomment the if ( ..) { } block below.

}

// Note: The handle_AsyncIcotl does get called when force_shutdown(void) gets

// called during Pegasus shutdown procedure (in case you ever wondered).

//if (_incoming_queue_shutdown.value() == 0 )

//{

// _shutdown_incoming_queue();

//}

_callback_ready.signal();

// _callback_thread.join();

_meta_dispatcher_mutex.lock(pegasus_thread_self());

{

AutoMutex autoMut(_meta_dispatcher_mutex);

_service_count--;

if (_service_count.value() == 0 )

{

_stop_polling++;

_polling_sem.signal();

_polling_thread->join();

Line 228

Line 265

_meta_dispatcher->_shutdown_routed_queue();

delete _meta_dispatcher;

_meta_dispatcher = 0;

delete _thread_pool;

_thread_pool = 0;

}

_meta_dispatcher_mutex.unlock();

} // mutex unlocks here

_polling_list.remove(this);

// Clean up in case there are extra stuff on the queue.

while (_incoming.count())

Line 243

Line 281

void MessageQueueService::_shutdown_incoming_queue(void)

{

if (_incoming_queue_shutdown.value() > 0 )

return ;

AsyncIoctl *msg = new AsyncIoctl(get_next_xid(),

Line 265

Line 304

_incoming.insert_last_wait(msg->op);

// _req_thread.join();

}

void MessageQueueService::enqueue(Message *msg) throw(IPCException)

void MessageQueueService::enqueue(Message *msg)

{

PEG_METHOD_ENTER(TRC_MESSAGEQUEUESERVICE, "MessageQueueService::enqueue()");

Line 467

Line 504

else

{

PEGASUS_ASSERT(rq != 0 );

// ATTN: optimization

// << Wed Mar 6 15:00:39 2002 mdd >>

// put thread and queue into the asyncopnode structure.

// (static_cast<AsyncMessage *>(rq))->_myself = operation->_thread_ptr;

// (static_cast<AsyncMessage *>(rq))->_service = operation->_service_ptr;

// done << Tue Mar 12 14:49:07 2002 mdd >>

operation->unlock();

_handle_async_request(static_cast<AsyncRequest *>(rq));

}

Line 612

Line 643

_polling_sem.signal();

return true;

}

// else

// {

// if( (rq != 0 && (true == MessageQueueService::messageOK(rq))) ||

// (rp != 0 && ( true == MessageQueueService::messageOK(rp) )) &&

// _die.value() == 0)

// {

// MessageQueueService::_incoming.insert_last_wait(op);

// return true;

// }

return false;

}

Line 633

Line 653

return true;

}

// made pure virtual

// << Wed Mar 6 15:11:31 2002 mdd >>

// void MessageQueueService::handleEnqueue(Message *msg)

// {

// if ( msg )

// delete msg;

// }

// made pure virtual

// << Wed Mar 6 15:11:56 2002 mdd >>

// void MessageQueueService::handleEnqueue(void)

// {

// Message *msg = dequeue();

// handleEnqueue(msg);

// }

void MessageQueueService::handle_heartbeat_request(AsyncRequest *req)

{

// default action is to echo a heartbeat response

Line 679

Line 682

{

case AsyncIoctl::IO_CLOSE:

{

// save my bearings

// Thread *myself = req->op->_thread_ptr;

MessageQueueService *service = static_cast<MessageQueueService *>(req->op->_service_ptr);

// respond to this message.

#ifdef MESSAGEQUEUESERVICE_DEBUG

PEGASUS_STD(cout) << service->getQueueName() << " Received AsyncIoctl::IO_CLOSE " << PEGASUS_STD(endl);

#endif

// respond to this message. this is fire and forget, so we don't need to delete anything.

// this takes care of two problems that were being found

// << Thu Oct 9 10:52:48 2003 mdd >>

_make_response(req, async_results::OK);

// ensure we do not accept any further messages

Line 707

Line 715

}

if( operation )

{

// operation->_thread_ptr = myself;

operation->_service_ptr = service;

service->_handle_incoming_operation(operation);

}

Line 717

Line 724

// shutdown the AsyncDQueue

service->_incoming.shutdown_queue();

AsyncOpNode *op = req->op;

op->_request.remove_first();

op->release();

return_op(op);

delete req;

// exit the thread !

// myself->exit_self( (PEGASUS_THREAD_RETURN) 1 );

return;

}

Line 734

void MessageQueueService::handle_CimServiceStart(CimServiceStart *req)

{

#ifdef MESSAGEQUEUESERVICE_DEBUG

PEGASUS_STD(cout) << getQueueName() << "received START" << PEGASUS_STD(endl);

#endif

// clear the stoped bit and update

_capabilities &= (~(module_capabilities::stopped));

_make_response(req, async_results::OK);

Line 743

Line 748

}

void MessageQueueService::handle_CimServiceStop(CimServiceStop *req)

{

#ifdef MESSAGEQUEUESERVICE_DEBUG

PEGASUS_STD(cout) << getQueueName() << "received STOP" << PEGASUS_STD(endl);

#endif

// set the stopeed bit and update

_capabilities |= module_capabilities::stopped;

_make_response(req, async_results::CIM_STOPPED);

Line 968

Line 976

Boolean destroy_op = false;

if (request->op == false)

if (request->op == 0)

{

request->op = get_op();

request->op->_request.insert_first(request);

Line 1165

Line 1173

Uint32 MessageQueueService::get_next_xid(void)

{

static Mutex _monitor;

Uint32 value;

AutoMutex autoMut(_monitor);

_xid++;

return _xid.value();

value = _xid.value();

return value;

}

PEGASUS_NAMESPACE_END

Legend:

Removed from v.1.70.2.1
changed lines
	Added in v.1.88

No CVS admin address has been configured