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.3 and 1.25.2.1

version 1.3, 2002/06/01 00:57:14

version 1.25.2.1, 2005/08/12 22:52:43

Line 1

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

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

Line 25

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

// Author: Mike Brasher (mbrasher@bmc.com)

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

// Modified By:

// Modified By: Dan Gorey (djgorey@us.ibm.com)

// Mike Day (mdday@us.ibm.com)s

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

// Nitin Upasani, Hewlett-Packard Company (Nitin_Upasani@hp.com)

// Nag Boranna, Hewlett-Packard Company (nagaraja_boranna@hp.com)

// Yi Zhou, Hewlett-Packard Company (yi_zhou@hp.com)

// Jenny Yu, Hewlett-Packard Company (jenny_yu@hp.com)

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

#include <Pegasus/Common/Config.h>

#include "CIMListener.h"

#include <iostream>

#include <Pegasus/Common/Exception.h>

#include <cassert>

#include <Pegasus/Common/SSLContext.h>

#include <cstdio>

#include <Pegasus/Common/Monitor.h>

#include <cctype>

#include <ctime>

#include <Pegasus/Common/FileSystem.h>

#include <Pegasus/Common/HTTPAcceptor.h>

#include <Pegasus/Common/Tracer.h>

#include <Pegasus/Common/PegasusVersion.h>

#include <Pegasus/Repository/CIMRepository.h>

#include <Pegasus/ExportServer/CIMExportRequestDispatcher.h>

#include <Pegasus/ExportServer/CIMExportResponseEncoder.h>

#include <Pegasus/ExportServer/CIMExportRequestDecoder.h>

#include "CIMListener.h"

#include <Pegasus/Consumer/CIMIndicationConsumer.h>

#include <Pegasus/Listener/CIMListenerIndicationDispatcher.h>

#define DDD(X) // X

PEGASUS_USING_STD;

PEGASUS_NAMESPACE_BEGIN

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

// CIMListenerService

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

class CIMListenerService

{

public:

CIMListenerService(Uint32 portNumber, SSLContext* sslContext=NULL);

CIMListenerService(CIMListenerService& svc);

~CIMListenerService();

void init();

/** bind to the port

void bind();

/** runForever Main runloop for the server.

void runForever();

/** Call to gracefully shutdown the server. The server connection socket

will be closed to disable new connections from clients.

void stopClientConnection();

/** Call to gracefully shutdown the server. It is called when the server

has been stopped and is ready to be shutdown. Next time runForever()

is called, the server shuts down.

void shutdown();

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

Boolean terminated() { return _dieNow; };

/** Call to resume the sever.

void resume();

/** Call to set the CIMServer state. Also inform the appropriate

message queues about the current state of the CIMServer.

void setState(Uint32 state);

Uint32 getOutstandingRequestCount();

/** Returns the indication listener dispatcher

CIMListenerIndicationDispatcher* getIndicationDispatcher() const;

/** Returns the indication listener dispatcher

void setIndicationDispatcher(CIMListenerIndicationDispatcher* dispatcher);

static PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL _listener_routine(void *param);

private:

Uint32 _portNumber;

SSLContext* _sslContext;

#ifdef PEGASUS_USE_23HTTPMONITOR_CLIENT

Monitor* _monitor;

HTTPAcceptor* _acceptor;

#else

monitor_2* _monitor;

pegasus_acceptor* _acceptor;

#endif

Boolean _dieNow;

CIMListenerIndicationDispatcher* _dispatcher;

CIMExportResponseEncoder* _responseEncoder;

CIMExportRequestDecoder* _requestDecoder;

};

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

:_portNumber(portNumber)

,_sslContext(sslContext)

,_monitor(NULL)

,_acceptor(NULL)

,_dieNow(false)

,_dispatcher(NULL)

,_responseEncoder(NULL)

,_requestDecoder(NULL)

{

}

CIMListener::CIMListener(

CIMListenerService::CIMListenerService(CIMListenerService& svc)

Monitor* monitor,

:_portNumber(svc._portNumber)

const String& rootPath,

,_sslContext(svc._sslContext)

Boolean dynamicReg,

,_monitor(NULL)

Boolean staticConsumers,

,_acceptor(NULL)

Boolean persistence)

,_dieNow(svc._dieNow)

: _dieNow(false), _rootPath(rootPath),

,_dispatcher(NULL)

_dynamicReg(dynamicReg),

,_responseEncoder(NULL)

_staticConsumers(staticConsumers),

,_requestDecoder(NULL)

_persistence(persistence)

{

const char METHOD_NAME[] = "CIMListener::CIMListener()";

}

CIMListenerService::~CIMListenerService()

{

// if port is alive, clean up the port

//if(_sslContext!=NULL)

// delete _sslContext;

PEG_FUNC_ENTER(TRC_SERVER, METHOD_NAME);

if(_responseEncoder!=NULL)

delete _responseEncoder;

// -- Save the monitor or create a new one:

if(_requestDecoder!=NULL)

delete _requestDecoder;

_monitor = monitor;

//if(_dispatcher!=NULL)

// delete _dispatcher;

// -- Create a CIMListenerState object:

if(_monitor!=NULL)

delete _monitor;

_cimExportRequestDispatcher

if(_acceptor!=NULL)

= new CIMExportRequestDispatcher(dynamicReg, staticConsumers, persistence);

delete _acceptor;

}

_cimExportResponseEncoder

void CIMListenerService::init()

= new CIMExportResponseEncoder;

{

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

_cimExportRequestDecoder = new CIMExportRequestDecoder(

#ifdef PEGASUS_USE_23HTTPMONITOR_CLIENT

_cimExportRequestDispatcher,

_monitor = new Monitor(true);

_cimExportResponseEncoder->getQueueId());

#else

_monitor = new monitor_2();

#endif

//_dispatcher = new CIMListenerIndicationDispatcher();

_responseEncoder = new CIMExportResponseEncoder();

_requestDecoder = new CIMExportRequestDecoder(

_dispatcher,

_responseEncoder->getQueueId());

#ifdef PEGASUS_USE_23HTTPMONITOR_CLIENT

_acceptor = new HTTPAcceptor(

_monitor,

_requestDecoder,

false,

_portNumber,

_sslContext,

false);

#else

_acceptor = new pegasus_acceptor(_monitor,

_requestDecoder,

false,

_portNumber,

_sslContext);

#endif

SSLContext * sslcontext = NULL;

bind();

_acceptor = new HTTPAcceptor(_monitor, _cimExportRequestDecoder, sslcontext);

PEG_METHOD_EXIT();

}

void CIMListenerService::bind()

{

if(_acceptor!=NULL)

{ // Bind to the port

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

PEG_FUNC_EXIT(TRC_SERVER, METHOD_NAME);

}

CIMListener::~CIMListener()

void CIMListenerService::runForever()

{

static int modulator = 0;

if(!_dieNow)

{

#ifdef PEGASUS_USE_23HTTPMONITOR_CLIENT

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

{

modulator++;

try

{

//MessageQueueService::_check_idle_flag = 1;

//MessageQueueService::_polling_sem.signal();

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

}

catch(...)

{

}

if (handleShutdownSignal)

{

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

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

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

handleShutdownSignal = false;

}

#else

_monitor->run();

#endif

}

void CIMListenerService::shutdown()

{

const char METHOD_NAME[] = "CIMListener::~CIMListener()";

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

_dieNow = true;

#ifdef PEGASUS_USE_23HTTPMONITOR_CLIENT

_monitor->tickle();

#endif

PEG_METHOD_EXIT();

}

PEG_FUNC_ENTER(TRC_SERVER, METHOD_NAME);

void CIMListenerService::resume()

{

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

// Note: do not delete the acceptor because it belongs to the Monitor

if(_acceptor!=NULL)

// which takes care of disposing of it.

_acceptor->reopenConnectionSocket();

PEG_FUNC_EXIT(TRC_SERVER, METHOD_NAME);

PEG_METHOD_EXIT();

}

void CIMListener::bind(Uint32 port)

void CIMListenerService::stopClientConnection()

{

const char METHOD_NAME[] = "CIMListener::bind()";

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

PEG_FUNC_ENTER(TRC_SERVER, METHOD_NAME);

// tell Monitor to stop listening for client connections

#ifdef PEGASUS_USE_23HTTPMONITOR_CLIENT

_monitor->stopListeningForConnections(true);

#else

_monitor->stop();

#endif

// not the best place to build the service url, but it works for now

// because the address string is accessible mdday

// Wait 150 milliseconds to allow time for the Monitor to stop

// listening for client connections.

// This wait time is the timeout value for the select() call

// in the Monitor's run() method (currently set to 100

// milliseconds) plus a delta of 50 milliseconds. The reason

// for the wait here is to make sure that the Monitor entries

// are updated before closing the connection sockets.

// pegasus_sleep(150); Not needed now due to the semaphore in the Monitor

_acceptor->bind(port);

if(_acceptor!=NULL)

_acceptor->closeConnectionSocket();

PEG_FUNC_EXIT(TRC_SERVER, METHOD_NAME);

PEG_METHOD_EXIT();

}

void CIMListener::runForever()

Uint32 CIMListenerService::getOutstandingRequestCount()

{

//ATTN: Do not add Trace code in this method.

return _acceptor->getOutstandingRequestCount();

if(!_dieNow)

_monitor->run(100);

}

void CIMListener::stopClientConnection()

CIMListenerIndicationDispatcher* CIMListenerService::getIndicationDispatcher() const

{

const char METHOD_NAME[] = "CIMListener::stopClientConnection()";

return _dispatcher;

}

void CIMListenerService::setIndicationDispatcher(CIMListenerIndicationDispatcher* dispatcher)

{

_dispatcher = dispatcher;

}

PEG_FUNC_ENTER(TRC_SERVER, METHOD_NAME);

PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL CIMListenerService::_listener_routine(void *param)

{

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

_acceptor->closeConnectionSocket();

//svc->init(); bug 1394

while(!svc->terminated())

{

#if defined(PEGASUS_PLATFORM_DARWIN_PPC_GNU)

pthread_testcancel();

#endif

svc->runForever();

}

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:

CIMListenerRep(Uint32 portNumber, SSLContext* sslContext=NULL);

~CIMListenerRep();

Uint32 getPortNumber() const;

SSLContext* getSSLContext() const;

void setSSLContext(SSLContext* sslContext);

void start();

void stop();

Boolean isAlive();

Boolean addConsumer(CIMIndicationConsumer* consumer);

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)

:_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 (_thread_pool != NULL)

{

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

}

if(_sslContext!=NULL)

delete _sslContext;

PEG_FUNC_EXIT(TRC_SERVER, METHOD_NAME);

if(_dispatcher!=NULL)

delete _dispatcher;

if(_thread_pool!=NULL)

delete _thread_pool;

if(_listener_sem!=NULL)

delete _listener_sem;

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

}

void CIMListener::shutdown()

Uint32 CIMListenerRep::getPortNumber() const

{

const char METHOD_NAME[] = "CIMListener::shutdown()";

return _portNumber;

}

PEG_FUNC_ENTER(TRC_SERVER, METHOD_NAME);

SSLContext* CIMListenerRep::getSSLContext() const

{

return _sslContext;

}

void CIMListenerRep::setSSLContext(SSLContext* sslContext)

{

if(_sslContext!=NULL)

delete _sslContext;

_dieNow = true;

_sslContext = sslContext;

}

void CIMListenerRep::start()

{

// spawn a thread to do this

if(_thread_pool==NULL)

{

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

try

{

// Try to initialize the service (bug 1394)

svc->setIndicationDispatcher(_dispatcher);

svc->init();

}

catch(...)

{

// Error. Exit without creating the ThreadPool, so that this listener

// is not 'alive'

delete svc;

throw;

}

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

create_time, destroy_time, deadlock_time);

PEG_FUNC_EXIT(TRC_SERVER, METHOD_NAME);

_listener_sem = new Semaphore(0);

if (_thread_pool->allocate_and_awaken(svc,

CIMListenerService::_listener_routine, _listener_sem) != PEGASUS_THREAD_OK)

{

Logger::put(Logger::STANDARD_LOG, System::CIMSERVER, Logger::TRACE,

"Not enough threads to start CIMListernerService.");

Tracer::trace(TRC_SERVER, Tracer::LEVEL2,

"Could not allocate thread for CIMListenerService::_listener_routine.");

throw Exception("Could not allocate thread.");

}

_svc = svc;

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

Logger::INFORMATION,

"CIMListener started");

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

}

void CIMListener::resume()

void CIMListenerRep::stop()

{

const char METHOD_NAME[] = "CIMListener::resume()";

if(_thread_pool!=NULL)

{

// Graceful shutdown of the listener service

PEG_FUNC_ENTER(TRC_SERVER, METHOD_NAME);

// Block incoming export requests and unbind the port

_svc->stopClientConnection();

_acceptor->reopenConnectionSocket();

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

PEG_FUNC_EXIT(TRC_SERVER, METHOD_NAME);

// 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::INFORMATION,

"CIMListener stopped");

}

CIMExportRequestDispatcher* CIMListener::getDispatcher()

Boolean CIMListenerRep::isAlive()

{

const char METHOD_NAME[] = "CIMListener::getDispatcher()";

return (_thread_pool!=NULL)?true:false;

}

PEG_FUNC_ENTER(TRC_SERVER, METHOD_NAME);

Boolean CIMListenerRep::addConsumer(CIMIndicationConsumer* consumer)

{

return _dispatcher->addConsumer(consumer);

}

Boolean CIMListenerRep::removeConsumer(CIMIndicationConsumer* consumer)

{

return _dispatcher->removeConsumer(consumer);

}

PEG_FUNC_EXIT(TRC_SERVER, METHOD_NAME);

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)

pegasus_sleep(100);

else

return true;

}

return _cimExportRequestDispatcher;

return false;

}

Uint32 CIMListener::getOutstandingRequestCount()

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

// CIMListener

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

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

:_rep(new CIMListenerRep(portNumber,sslContext))

{

}

CIMListener::~CIMListener()

{

const char METHOD_NAME[] = "CIMListener::getOutstandingRequestCount()";

if(_rep!=NULL)

delete static_cast<CIMListenerRep*>(_rep);

_rep=NULL;

}

PEG_FUNC_ENTER(TRC_SERVER, METHOD_NAME);

Uint32 CIMListener::getPortNumber() const

{

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

}

PEG_FUNC_EXIT(TRC_SERVER, METHOD_NAME);

SSLContext* CIMListener::getSSLContext() const

{

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

{

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

}

return (_acceptor->getOutstandingRequestCount());

Boolean CIMListener::addConsumer(CIMIndicationConsumer* consumer)

{

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

}

Boolean CIMListener::removeConsumer(CIMIndicationConsumer* consumer)

{

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

}

PEGASUS_NAMESPACE_END

Legend:

Removed from v.1.3
changed lines
	Added in v.1.25.2.1

No CVS admin address has been configured