pegasus/src/Pegasus/Listener/CIMListener.cpp - diff

Return to CIMListener.cpp CVS log

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

Diff for /pegasus/src/Pegasus/Listener/CIMListener.cpp between version 1.9 and 1.47.6.3

version 1.9, 2003/08/16 14:52:27

version 1.47.6.3, 2007/06/22 10:46:13

Line 1

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

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

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

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

Line 29

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

// Author: Dong Xiang, EMC Corporation (xiang_dong@emc.com)

// Modified By:

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

#include "CIMListener.h"

#include <Pegasus/Common/Exception.h>

#include <Pegasus/Common/SSLContext.h>

#include <Pegasus/Common/Monitor.h>

#include <Pegasus/Common/HTTPAcceptor.h>

#include <Pegasus/Common/PegasusVersion.h>

#include <Pegasus/Common/MessageLoader.h>

#include <Pegasus/Common/Time.h>

#include <Pegasus/ExportServer/CIMExportResponseEncoder.h>

#include <Pegasus/ExportServer/CIMExportRequestDecoder.h>

#include <Pegasus/Consumer/CIMIndicationConsumer.h>

#include <Pegasus/Listener/CIMListenerIndicationDispatcher.h>

PEGASUS_NAMESPACE_BEGIN

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

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

// CIMListenerService

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

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

class CIMListenerService

{

public:

Line 52

Line 60

~CIMListenerService();

void init();

/** bind to the port

void bind();

/** runForever Main runloop for the server.

void runForever();

Line 72

Line 82

/** Return true if the server has shutdown, false otherwise.

Boolean terminated() { return _dieNow; };

Boolean terminated() const

{

return _dieNow;

};

/** Call to resume the sever.

Line 93

Line 106

void setIndicationDispatcher(CIMListenerIndicationDispatcher* dispatcher);

static PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL _listener_routine(void *param);

/** Returns the port number being used.

Uint32 getPortNumber() const;

static ThreadReturnType PEGASUS_THREAD_CDECL

_listener_routine(void *param);

private:

Uint32 _portNumber;

SSLContext* _sslContext;

Monitor* _monitor;

HTTPAcceptor* _acceptor;

Mutex _monitorMutex;

#ifdef PEGASUS_ENABLE_IPV6

HTTPAcceptor *_ip6Acceptor;

#endif

#if !defined (PEGASUS_ENABLE_IPV6) || defined (PEGASUS_OS_TYPE_WINDOWS)

HTTPAcceptor *_ip4Acceptor;

#endif

Boolean _dieNow;

CIMListenerIndicationDispatcher* _dispatcher;

CIMExportResponseEncoder* _responseEncoder;

CIMExportRequestDecoder* _requestDecoder;

};

CIMListenerService::CIMListenerService(Uint32 portNumber, SSLContext* sslContext)

CIMListenerService::CIMListenerService(

:_portNumber(portNumber)

Uint32 portNumber,

,_sslContext(sslContext)

SSLContext * sslContext)

,_monitor(NULL)

,_acceptor(NULL)

_portNumber(portNumber),

,_dieNow(false)

_sslContext(sslContext),

,_dispatcher(NULL)

_monitor(NULL),

,_responseEncoder(NULL)

#ifdef PEGASUS_ENABLE_IPV6

,_requestDecoder(NULL)

_ip6Acceptor(NULL),

#endif

#if !defined (PEGASUS_ENABLE_IPV6) || defined (PEGASUS_OS_TYPE_WINDOWS)

_ip4Acceptor(NULL),

#endif

_dieNow(false),

_dispatcher(NULL),

_responseEncoder(NULL),

_requestDecoder(NULL)

{

}

CIMListenerService::CIMListenerService(CIMListenerService& svc)

CIMListenerService::CIMListenerService(CIMListenerService & svc) :

:_portNumber(svc._portNumber)

_portNumber(svc._portNumber),

,_sslContext(svc._sslContext)

_sslContext(svc._sslContext),

,_monitor(NULL)

_monitor(NULL),

,_acceptor(NULL)

#ifdef PEGASUS_ENABLE_IPV6

,_dieNow(svc._dieNow)

_ip6Acceptor(NULL),

,_dispatcher(NULL)

#endif

,_responseEncoder(NULL)

#if !defined (PEGASUS_ENABLE_IPV6) || defined (PEGASUS_OS_TYPE_WINDOWS)

,_requestDecoder(NULL)

_ip4Acceptor(NULL),

#endif

_dieNow(svc._dieNow),

_dispatcher(NULL),

_responseEncoder(NULL),

_requestDecoder(NULL)

{

}

CIMListenerService::~CIMListenerService()

{

// if port is alive, clean up the port

//if(_sslContext!=NULL)

// delete _sslContext;

if(_responseEncoder!=NULL)

delete _responseEncoder;

if(_requestDecoder!=NULL)

delete _requestDecoder;

#ifdef PEGASUS_ENABLE_IPV6

//if(_dispatcher!=NULL)

delete _ip6Acceptor;

// delete _dispatcher;

#endif

#if !defined (PEGASUS_ENABLE_IPV6) || defined (PEGASUS_OS_TYPE_WINDOWS)

if(_monitor!=NULL)

delete _ip4Acceptor;

#endif

delete _monitor;

if(_acceptor!=NULL)

delete _acceptor;

}

void CIMListenerService::init()

{

PEG_METHOD_ENTER(TRC_LISTENER, "CIMListenerService::init");

_monitor = new Monitor(true);

if (NULL == _monitor)

_monitor = new Monitor();

//_dispatcher = new CIMListenerIndicationDispatcher();

if (NULL == _responseEncoder)

_responseEncoder = new CIMExportResponseEncoder();

_requestDecoder = new CIMExportRequestDecoder(

_dispatcher,

_responseEncoder->getQueueId());

_acceptor = new HTTPAcceptor(

_monitor,

_requestDecoder,

false,

_portNumber,

_sslContext);

if (NULL == _requestDecoder)

{

_requestDecoder = new CIMExportRequestDecoder(

_dispatcher, _responseEncoder->getQueueId());

}

#ifdef PEGASUS_ENABLE_IPV6

if (NULL == _ip6Acceptor)

{

_ip6Acceptor = new HTTPAcceptor(

_monitor, _requestDecoder, HTTPAcceptor::IPV6_CONNECTION,

_portNumber, _sslContext, false);

}

#endif

#if !defined (PEGASUS_ENABLE_IPV6) || defined (PEGASUS_OS_TYPE_WINDOWS)

if (NULL == _ip4Acceptor)

{

_ip4Acceptor = new HTTPAcceptor(

_monitor, _requestDecoder, HTTPAcceptor::IPV4_CONNECTION,

_portNumber, _sslContext, false);

}

#endif

bind();

PEG_METHOD_EXIT();

}

void CIMListenerService::bind()

{

if(_acceptor!=NULL)

#ifdef PEGASUS_ENABLE_IPV6

{ // Bind to the port

if (_ip6Acceptor != NULL)

_acceptor->bind();

{

_ip6Acceptor->bind();

PEGASUS_STD(cout) << "Listening on HTTP port " << _portNumber << PEGASUS_STD(endl);

//listener.addAcceptor(false, portNumberHttp, false);

Logger::put(Logger::STANDARD_LOG, System::CIMLISTENER, Logger::INFORMATION,

"Listening on HTTP port $0.", _portNumber);

Logger::put(

Logger::STANDARD_LOG,

System::CIMLISTENER,

Logger::INFORMATION,

"IPV6, Listening on HTTP port $0.",

_portNumber);

}

#endif

#if !defined (PEGASUS_ENABLE_IPV6) || defined (PEGASUS_OS_TYPE_WINDOWS)

if (_ip4Acceptor != NULL)

{

_ip4Acceptor->bind();

Logger::put(

Logger::STANDARD_LOG,

System::CIMLISTENER,

Logger::INFORMATION,

"IPV4, Listening on HTTP for port $0.",

_portNumber);

}

#endif

}

void CIMListenerService::runForever()

{

static int modulator = 0;

if(!_dieNow)

{

if(false == _monitor->run(100))

_monitor->run(500000);

{

static struct timeval lastIdleCleanupTime = {0, 0};

modulator++;

struct timeval now;

if(!(modulator % 5000) )

Time::gettimeofday(&now);

if (now.tv_sec - lastIdleCleanupTime.tv_sec > 300)

{

lastIdleCleanupTime.tv_sec = now.tv_sec;

try

{

//MessageQueueService::_check_idle_flag = 1;

MessageQueueService::get_thread_pool()->cleanupIdleThreads();

//MessageQueueService::_polling_sem.signal();

_monitor->kill_idle_threads();

}

catch(...)

{

// Ignore!

}

if (handleShutdownSignal)

{

Tracer::trace(TRC_SERVER, Tracer::LEVEL3,

"CIMServer::runForever - signal received. Shutting down.");

ShutdownService::getInstance(this)->shutdown(true, 10, false);

handleShutdownSignal = false;

}

void CIMListenerService::shutdown()

{

PEG_METHOD_ENTER(TRC_LISTENER, "CIMListenerService::shutdown()");

// This logic signals the thread currently executing _listener_routine()

// to exit. That function deletes this instance of CIMListenerService,

// which deletes the _monitor member. We use a mutex to keep it from

// deleting the monitor until after tickle has been called.

{

AutoMutex am(_monitorMutex);

_dieNow = true;

_monitor->tickle();

}

PEG_METHOD_EXIT();

}

Line 239

Line 291

void CIMListenerService::resume()

{

PEG_METHOD_ENTER(TRC_LISTENER, "CIMListenerService::resume()");

#ifdef PEGASUS_ENABLE_IPV6

if(_acceptor!=NULL)

if (_ip6Acceptor != NULL)

_acceptor->reopenConnectionSocket();

{

_ip6Acceptor->reopenConnectionSocket();

}

#endif

#if !defined (PEGASUS_ENABLE_IPV6) || defined (PEGASUS_OS_TYPE_WINDOWS)

if (_ip4Acceptor != NULL)

{

_ip4Acceptor->reopenConnectionSocket();

}

#endif

PEG_METHOD_EXIT();

}

void CIMListenerService::stopClientConnection()

{

PEG_METHOD_ENTER(TRC_LISTENER, "CIMListenerService::stopClientConnection()");

PEG_METHOD_ENTER(

TRC_LISTENER,

if(_acceptor!=NULL)

"CIMListenerService::stopClientConnection()");

_acceptor->closeConnectionSocket();

// tell Monitor to stop listening for client connections

_monitor->stopListeningForConnections(true);

#ifdef PEGASUS_ENABLE_IPV6

if (_ip6Acceptor != NULL)

{

_ip6Acceptor->closeConnectionSocket();

}

#endif

#if !defined (PEGASUS_ENABLE_IPV6) || defined (PEGASUS_OS_TYPE_WINDOWS)

if (_ip4Acceptor != NULL)

{

_ip4Acceptor->closeConnectionSocket();

}

#endif

PEG_METHOD_EXIT();

}

Uint32 CIMListenerService::getOutstandingRequestCount()

{

Uint32 cnt = 0;

#ifdef PEGASUS_ENABLE_IPV6

cnt = _ip6Acceptor->getOutstandingRequestCount();

#endif

#if !defined (PEGASUS_ENABLE_IPV6) || defined (PEGASUS_OS_TYPE_WINDOWS)

cnt += _ip4Acceptor->getOutstandingRequestCount();

#endif

return cnt;

}

CIMListenerIndicationDispatcher* CIMListenerService::getIndicationDispatcher() const

CIMListenerIndicationDispatcher*

CIMListenerService::getIndicationDispatcher() const

{

return _dispatcher;

}

void CIMListenerService::setIndicationDispatcher(CIMListenerIndicationDispatcher* dispatcher)

void CIMListenerService::setIndicationDispatcher(

CIMListenerIndicationDispatcher* dispatcher)

{

_dispatcher = dispatcher;

}

PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL CIMListenerService::_listener_routine(void *param)

Uint32 CIMListenerService::getPortNumber() const

{

Uint32 portNumber = _portNumber;

#ifdef PEGASUS_ENABLE_IPV6

if ((portNumber == 0) && (_ip6Acceptor != 0))

{

portNumber = _ip6Acceptor->getPortNumber();

}

#endif

#if !defined (PEGASUS_ENABLE_IPV6) || defined (PEGASUS_OS_TYPE_WINDOWS)

if ((portNumber == 0) && (_ip4Acceptor != 0))

{

portNumber = _ip4Acceptor->getPortNumber();

}

#endif

return (portNumber);

}

ThreadReturnType PEGASUS_THREAD_CDECL

CIMListenerService::_listener_routine(void *param)

{

CIMListenerService *svc = reinterpret_cast<CIMListenerService *>(param);

svc->init();

try

{

// svc->init(); bug 1394

while(!svc->terminated())

{

#if defined(PEGASUS_OS_DARWIN)

pthread_testcancel();

#endif

svc->runForever();

}

catch(...)

{

PEG_TRACE_CSTRING(TRC_SERVER, Tracer::LEVEL2,

"Unknown exception thrown in _listener_routine.");

}

// CAUTION: deleting the service also deletes the monitor whose tickle()

// method may still be executing in another thread. This line of code was

// most likely reached when the CIMListenerService::shutdown() method set

// _dieNow to true and called Monitor::tickle(). We must wait until we

// can obtain the _monitorMutex, indicating that we are no longer inside

// Monitor::tickle().

svc->_monitorMutex.lock();

svc->_monitorMutex.unlock();

delete svc;

return 0;

}

static struct timeval create_time = {0, 1};

static struct timeval destroy_time = {15, 0};

static struct timeval deadlock_time = {0, 0};

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

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

// CIMListenerRep

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

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

class CIMListenerRep

{

public:

Line 306

Line 436

Boolean removeConsumer(CIMIndicationConsumer* consumer);

private:

Boolean waitForPendingRequests(Uint32 shutdownTimeout);

Uint32 _portNumber;

SSLContext* _sslContext;

CIMListenerIndicationDispatcher* _dispatcher;

ThreadPool* _thread_pool;

CIMListenerService *_svc;

Semaphore *_listener_sem;

};

CIMListenerRep::CIMListenerRep(Uint32 portNumber, SSLContext* sslContext)

CIMListenerRep::CIMListenerRep(

:_portNumber(portNumber)

Uint32 portNumber,

,_sslContext(sslContext)

SSLContext * sslContext)

,_dispatcher(new CIMListenerIndicationDispatcher())

,_thread_pool(NULL)

_portNumber(portNumber),

_sslContext(sslContext),

_dispatcher(new CIMListenerIndicationDispatcher()),

_thread_pool(NULL),

_svc(NULL),

_listener_sem(NULL)

{

}

CIMListenerRep::~CIMListenerRep()

{

// if port is alive, clean up the port

if(_sslContext!=NULL)

if (_thread_pool != 0)

delete _sslContext;

{

// Block incoming export requests and unbind the port

_svc->stopClientConnection();

if(_dispatcher!=NULL)

// Wait until pending export requests in the server are done.

delete _dispatcher;

waitForPendingRequests(10);

if(_thread_pool!=NULL)

// Shutdown the CIMListenerService

_svc->shutdown();

}

delete _sslContext;

delete _dispatcher;

delete _thread_pool;

delete _listener_sem;

// don't delete _svc, this is deleted by _listener_routine

}

Uint32 CIMListenerRep::getPortNumber() const

{

return _portNumber;

Uint32 portNumber;

if (_svc == 0)

portNumber = _portNumber;

else

portNumber = _svc->getPortNumber();

return portNumber;

}

SSLContext* CIMListenerRep::getSSLContext() const

{

return _sslContext;

}

void CIMListenerRep::setSSLContext(SSLContext* sslContext)

{

if(_sslContext!=NULL)

delete _sslContext;

_sslContext = sslContext;

}

void CIMListenerRep::start()

{

// spawn a thread to do this

if(_thread_pool==NULL)

if (_thread_pool == 0)

{

_thread_pool = new ThreadPool(0, "Listener", 0, 1,

AutoPtr < CIMListenerService >

create_time, destroy_time, deadlock_time);

svc(new CIMListenerService(_portNumber, _sslContext));

CIMListenerService* svc = new CIMListenerService(_portNumber,_sslContext);

svc->setIndicationDispatcher(_dispatcher);

svc->init();

_thread_pool->allocate_and_awaken(svc,CIMListenerService::_listener_routine);

struct timeval deallocateWait = { 15, 0 };

AutoPtr < ThreadPool >

Logger::put(Logger::STANDARD_LOG,System::CIMLISTENER,

threadPool(new ThreadPool(0, "Listener", 0, 1, deallocateWait));

AutoPtr < Semaphore > sem(new Semaphore(0));

if (threadPool->allocate_and_awaken(

svc.get(), CIMListenerService::_listener_routine, sem.get())

!= PEGASUS_THREAD_OK)

{

Logger::put(

Logger::STANDARD_LOG, System::CIMLISTENER,

Logger::TRACE,

"Not enough threads to start CIMListernerService.");

PEG_TRACE_CSTRING(

TRC_SERVER,

Tracer::LEVEL2,

"Could not allocate thread for "

"CIMListenerService::_listener_routine.");

throw

Exception(MessageLoaderParms(

"Listener.CIMListener.CANNOT_ALLOCATE_THREAD",

"Could not allocate thread."));

}

Logger::put(

Logger::STANDARD_LOG,

System::CIMLISTENER,

Logger::INFORMATION,

"CIMListener started");

PEGASUS_STD(cerr) << "CIMlistener started" << PEGASUS_STD(endl);

_svc = svc.release();

_thread_pool = threadPool.release();

_listener_sem = sem.release();

}

void CIMListenerRep::stop()

{

if(_thread_pool!=NULL)

{ // stop the thread

{

// Graceful shutdown of the listener service

// Block incoming export requests and unbind the port

_svc->stopClientConnection();

// Wait until pending export requests in the server are done.

waitForPendingRequests(10);

// Shutdown the CIMListenerService

_svc->shutdown();

// Wait for the _listener_routine thread to exit.

// The thread could be delivering an export, so give it 3sec.

// Note that _listener_routine deletes the CIMListenerService,

// so no need to delete _svc.

try

{

_listener_sem->time_wait(3000);

}

catch(const TimeOut &)

{

// No need to do anything, the thread pool will be deleted below

// to cancel the _listener_routine thread if it is still running.

}

delete _listener_sem;

_listener_sem = NULL;

// Delete the thread pool. This cancels the listener thread if it is

// still

// running.

delete _thread_pool;

_thread_pool = NULL;

Logger::put(Logger::STANDARD_LOG,System::CIMLISTENER,

Logger::put(

Logger::INFORMATION,

Logger::STANDARD_LOG, System::CIMLISTENER,

"CIMListener stopped");

Logger::INFORMATION, "CIMListener stopped");

}

Line 392

Line 609

{

return _dispatcher->addConsumer(consumer);

}

Boolean CIMListenerRep::removeConsumer(CIMIndicationConsumer* consumer)

{

return _dispatcher->removeConsumer(consumer);

}

Boolean CIMListenerRep::waitForPendingRequests(Uint32 shutdownTimeout)

{

// Wait for 10 sec max

Uint32 reqCount;

Uint32 countDown = shutdownTimeout * 10;

for (; countDown > 0; countDown--)

{

reqCount = _svc->getOutstandingRequestCount();

if (reqCount > 0)

Threads::sleep(100);

else

return true;

}

return false;

}

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

// CIMListener

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

CIMListener::CIMListener(Uint32 portNumber, SSLContext* sslContext)

:_rep(new CIMListenerRep(portNumber,sslContext))

CIMListener::CIMListener(

Uint32 portNumber,

SSLContext * sslContext)

_rep(new CIMListenerRep(portNumber, sslContext))

{

}

CIMListener::~CIMListener()

{

if(_rep!=NULL)

Line 420

Line 663

{

return static_cast<CIMListenerRep*>(_rep)->getSSLContext();

}

void CIMListener::setSSLContext(SSLContext* sslContext)

{

static_cast<CIMListenerRep*>(_rep)->setSSLContext(sslContext);

}

void CIMListener::start()

{

static_cast<CIMListenerRep*>(_rep)->start();

}

void CIMListener::stop()

{

static_cast<CIMListenerRep*>(_rep)->stop();

}

Boolean CIMListener::isAlive()

Boolean CIMListener::isAlive() const

{

return static_cast<CIMListenerRep*>(_rep)->isAlive();

}

Line 442

Line 688

{

return static_cast<CIMListenerRep*>(_rep)->addConsumer(consumer);

}

Boolean CIMListener::removeConsumer(CIMIndicationConsumer* consumer)

{

return static_cast<CIMListenerRep*>(_rep)->removeConsumer(consumer);

Legend:

Removed from v.1.9
changed lines
	Added in v.1.47.6.3

No CVS admin address has been configured