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

Return to Monitor.cpp CVS log

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

Diff for /pegasus/src/Pegasus/Common/Monitor.cpp between version 1.70 and 1.135

version 1.70, 2004/01/30 15:41:28

version 1.135, 2008/06/20 17:20:35

Line 1

//%2003////////////////////////////////////////////////////////////////////////

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

// Company, L. P., IBM Corp., 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 23

Line 29

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

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

// Modified By: Mike Day (monitor_2) mdday@us.ibm.com

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

#include "Network.h"

#include <Pegasus/Common/Config.h>

#include <cstring>

#include "Monitor.h"

#include "MessageQueue.h"

#include "Socket.h"

#include <Pegasus/Common/Tracer.h>

#include <Pegasus/Common/HTTPConnection.h>

#include <Pegasus/Common/HTTPAcceptor.h>

#include <Pegasus/Common/MessageQueueService.h>

#include <Pegasus/Common/Exception.h>

#ifdef PEGASUS_OS_TYPE_WINDOWS

#include "ArrayIterator.h"

# if defined(FD_SETSIZE) && FD_SETSIZE != 1024

#include "HostAddress.h"

# error "FD_SETSIZE was not set to 1024 prior to the last inclusion \

#include <errno.h>

of <winsock.h>. It may have been indirectly included (e.g., by including \

<windows.h>). Finthe inclusion of that header which is visible to this \

compilation unit and #define FD_SETZIE to 1024 prior to that inclusion; \

otherwise, less than 64 clients (the default) will be able to connect to the \

CIMOM. PLEASE DO NOT SUPPRESS THIS WARNING; PLEASE FIX THE PROBLEM."

# endif

# define FD_SETSIZE 1024

# include <windows.h>

#else

# include <sys/types.h>

# include <sys/socket.h>

# include <sys/time.h>

# include <netinet/in.h>

# include <netdb.h>

# include <arpa/inet.h>

#endif

PEGASUS_USING_STD;

PEGASUS_NAMESPACE_BEGIN

static AtomicInt _connections = 0;

static struct timeval create_time = {0, 1};

static struct timeval destroy_time = {300, 0};

static struct timeval deadlock_time = {0, 0};

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

// MonitorRep

// Tickler

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

struct MonitorRep

Tickler::Tickler()

: _listenSocket(PEGASUS_INVALID_SOCKET),

_clientSocket(PEGASUS_INVALID_SOCKET),

_serverSocket(PEGASUS_INVALID_SOCKET)

{

fd_set rd_fd_set;

try

fd_set wr_fd_set;

fd_set ex_fd_set;

fd_set active_rd_fd_set;

fd_set active_wr_fd_set;

fd_set active_ex_fd_set;

};

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

// Monitor

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

#define MAX_NUMBER_OF_MONITOR_ENTRIES 32

Monitor::Monitor()

: _module_handle(0), _controller(0), _async(false), _stopConnections(0)

{

int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;

Socket::initializeInterface();

_rep = 0;

_entries.reserveCapacity(numberOfMonitorEntriesToAllocate);

for( int i = 0; i < numberOfMonitorEntriesToAllocate; i++ )

{

_MonitorEntry entry(0, 0, 0);

_initialize();

_entries.append(entry);

}

catch (...)

Monitor::Monitor(Boolean async)

: _module_handle(0), _controller(0), _async(async), _stopConnections(0)

{

int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;

Socket::initializeInterface();

_rep = 0;

_entries.reserveCapacity(numberOfMonitorEntriesToAllocate);

for( int i = 0; i < numberOfMonitorEntriesToAllocate; i++ )

{

_MonitorEntry entry(0, 0, 0);

_uninitialize();

_entries.append(entry);

throw;

}

Monitor::~Monitor()

Tickler::~Tickler()

{

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

"deregistering with module controller");

if(_module_handle != NULL)

{

_controller->deregister_module(PEGASUS_MODULENAME_MONITOR);

_uninitialize();

_controller = 0;

delete _module_handle;

}

Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "deleting rep");

Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");

Socket::uninitializeInterface();

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

"returning from monitor destructor");

}

Boolean Monitor::run(Uint32 milliseconds)

void Tickler::notify()

{

Socket::write(_clientSocket, "\0", 1);

Boolean handled_events = false;

int i = 0;

#if defined(PEGASUS_OS_OS400) || defined(PEGASUS_OS_HPUX)

struct timeval tv = {milliseconds/1000, milliseconds%1000*1000};

#else

struct timeval tv = {0, 1};

#endif

fd_set fdread;

FD_ZERO(&fdread);

_entry_mut.lock(pegasus_thread_self());

// Check the stopConnections flag. If set, clear the Acceptor monitor entries

if (_stopConnections == 1)

{

for ( int indx = 0; indx < (int)_entries.size(); indx++)

{

if (_entries[indx]._type == Monitor::ACCEPTOR)

{

if ( _entries[indx]._status.value() != _MonitorEntry::EMPTY)

{

if ( _entries[indx]._status.value() == _MonitorEntry::IDLE ||

_entries[indx]._status.value() == _MonitorEntry::DYING )

{

// remove the entry

_entries[indx]._status = _MonitorEntry::EMPTY;

}

else

{

// set status to DYING

_entries[indx]._status = _MonitorEntry::DYING;

}

_stopConnections = 0;

}

for( int indx = 0; indx < (int)_entries.size(); indx++)

void Tickler::reset()

{

if ((_entries[indx]._status.value() == _MonitorEntry::DYING) &&

// Clear all bytes from the tickle socket

(_entries[indx]._type == Monitor::CONNECTION))

char buffer[32];

while (Socket::read(_serverSocket, buffer, 32) > 0)

{

MessageQueue *q = MessageQueue::lookup(_entries[indx].queueId);

PEGASUS_ASSERT(q != 0);

MessageQueue & o = static_cast<HTTPConnection *>(q)->get_owner();

Message* message= new CloseConnectionMessage(_entries[indx].socket);

message->dest = o.getQueueId();

// HTTPAcceptor is responsible for closing the connection.

// The lock is released to allow HTTPAcceptor to call

// unsolicitSocketMessages to free the entry.

// Once HTTPAcceptor completes processing of the close

// connection, the lock is re-requested and processing of

// the for loop continues. This is safe with the current

// implementation of the _entries object. Note that the

// loop condition accesses the _entries.size() on each

// iteration, so that a change in size while the mutex is

// unlocked will not result in an ArrayIndexOutOfBounds

// exception.

_entry_mut.unlock();

o.enqueue(message);

_entry_mut.lock(pegasus_thread_self());

}

Uint32 _idleEntries = 0;

#if defined(PEGASUS_OS_TYPE_UNIX)

for( int indx = 0; indx < (int)_entries.size(); indx++)

// Use an anonymous pipe for the tickle connection.

{

if(_entries[indx]._status.value() == _MonitorEntry::IDLE)

{

_idleEntries++;

FD_SET(_entries[indx].socket, &fdread);

}

// Fixed in monitor_2 but added because Monitor is still the default monitor.

void Tickler::_initialize()

// When _idleEntries is 0 don't immediately return, otherwise this loops out of control

// kicking off kill idle thread threads. E.g. There is nothing to select on when the cimserver

// is shutting down.

if( _idleEntries == 0 )

{

Thread::sleep( milliseconds );

int fds[2];

_entry_mut.unlock();

return false;

}

_entry_mut.unlock();

int events = select(FD_SETSIZE, &fdread, NULL, NULL, &tv);

_entry_mut.lock(pegasus_thread_self());

#ifdef PEGASUS_OS_TYPE_WINDOWS

if (pipe(fds) == -1)

if(events == SOCKET_ERROR)

#else

if(events == -1)

#endif

{

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

MessageLoaderParms parms(

"Monitor::run - errorno = %d has occurred on select.", errno);

"Common.Monitor.TICKLE_CREATE",

// The EBADF error indicates that one or more or the file

"Received error number $0 while creating the internal socket.",

// descriptions was not valid. This could indicate that

getSocketError());

// the _entries structure has been corrupted or that

throw Exception(parms);

// we have a synchronization error.

PEGASUS_ASSERT(errno != EBADF);

}

else if (events)

{

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

"Monitor::run select event received events = %d, monitoring %d idle entries",

events, _idleEntries);

for( int indx = 0; indx < (int)_entries.size(); indx++)

{

// The Monitor should only look at entries in the table that are IDLE (i.e.,

// owned by the Monitor).

if((_entries[indx]._status.value() == _MonitorEntry::IDLE) &&

(FD_ISSET(_entries[indx].socket, &fdread)))

{

MessageQueue *q = MessageQueue::lookup(_entries[indx].queueId);

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

"Monitor::run indx = %d, queueId = %d, q = %p",

indx, _entries[indx].queueId, q);

PEGASUS_ASSERT(q !=0);

try

_serverSocket = fds[0];

{

_clientSocket = fds[1];

if(_entries[indx]._type == Monitor::CONNECTION)

{

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

"_entries[indx].type for indx = %d is Monitor::CONNECTION", indx);

static_cast<HTTPConnection *>(q)->_entry_index = indx;

_entries[indx]._status = _MonitorEntry::BUSY;

// If allocate_and_awaken failure, retry on next iteration

if (!MessageQueueService::get_thread_pool()->allocate_and_awaken(

(void *)q, _dispatch))

{

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

"Monitor::run: Insufficient resources to process request.");

_entries[indx]._status = _MonitorEntry::IDLE;

_entry_mut.unlock();

return true;

}

else

{

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

"Non-connection entry, indx = %d, has been received.", indx);

int events = 0;

events |= SocketMessage::READ;

Message *msg = new SocketMessage(_entries[indx].socket, events);

_entries[indx]._status = _MonitorEntry::BUSY;

_entry_mut.unlock();

q->enqueue(msg);

Socket::disableBlocking(_serverSocket);

_entries[indx]._status = _MonitorEntry::IDLE;

return true;

}

catch(...)

{

}

handled_events = true;

}

_entry_mut.unlock();

return(handled_events);

}

void Monitor::stopListeningForConnections()

#else

{

PEG_METHOD_ENTER(TRC_HTTP, "Monitor::stopListeningForConnections()");

_stopConnections = 1;

// Use an external loopback socket connection to allow the tickle socket to

// be included in the select() array on non-Unix platforms.

PEG_METHOD_EXIT();

void Tickler::_initialize()

}

{

// Set up the addresses for the listen, client, and server sockets

// based on whether IPv6 is enabled.

Socket::initializeInterface();

int Monitor::solicitSocketMessages(

# ifdef PEGASUS_ENABLE_IPV6

Sint32 socket,

struct sockaddr_storage listenAddress;

Uint32 events,

struct sockaddr_storage clientAddress;

Uint32 queueId,

struct sockaddr_storage serverAddress;

int type)

# else

{

struct sockaddr_in listenAddress;

struct sockaddr_in clientAddress;

struct sockaddr_in serverAddress;

# endif

PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitSocketMessages");

int addressFamily;

SocketLength addressLength;

_entry_mut.lock(pegasus_thread_self());

memset(&listenAddress, 0, sizeof (listenAddress));

for(int index = 0; index < (int)_entries.size(); index++)

# ifdef PEGASUS_ENABLE_IPV6

{

if (System::isIPv6StackActive())

try

{

if(_entries[index]._status.value() == _MonitorEntry::EMPTY)

// Use the IPv6 loopback address for the listen sockets

{

HostAddress::convertTextToBinary(

_entries[index].socket = socket;

HostAddress::AT_IPV6,

_entries[index].queueId = queueId;

"::1",

_entries[index]._type = type;

&reinterpret_cast<struct sockaddr_in6*>(&listenAddress)->sin6_addr);

_entries[index]._status = _MonitorEntry::IDLE;

listenAddress.ss_family = AF_INET6;

_entry_mut.unlock();

reinterpret_cast<struct sockaddr_in6*>(&listenAddress)->sin6_port = 0;

return index;

addressFamily = AF_INET6;

}

addressLength = sizeof(struct sockaddr_in6);

}

catch(...)

else

# endif

{

}

// Use the IPv4 loopback address for the listen sockets

HostAddress::convertTextToBinary(

HostAddress::AT_IPV4,

"127.0.0.1",

&reinterpret_cast<struct sockaddr_in*>(

&listenAddress)->sin_addr.s_addr);

reinterpret_cast<struct sockaddr_in*>(&listenAddress)->sin_family =

AF_INET;

reinterpret_cast<struct sockaddr_in*>(&listenAddress)->sin_port = 0;

}

addressFamily = AF_INET;

_entry_mut.unlock();

addressLength = sizeof(struct sockaddr_in);

PEG_METHOD_EXIT();

return -1;

}

void Monitor::unsolicitSocketMessages(Sint32 socket)

// Use the same address for the client socket as the listen socket

{

clientAddress = listenAddress;

PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");

_entry_mut.lock(pegasus_thread_self());

// Set up a listen socket to allow the tickle client and server to connect

for(int index = 0; index < (int)_entries.size(); index++)

// Create the listen socket

{

if ((_listenSocket = Socket::createSocket(addressFamily, SOCK_STREAM, 0)) ==

if(_entries[index].socket == socket)

PEGASUS_INVALID_SOCKET)

{

_entries[index]._status = _MonitorEntry::EMPTY;

MessageLoaderParms parms(

_entries[index].socket = -1;

"Common.Monitor.TICKLE_CREATE",

break;

"Received error number $0 while creating the internal socket.",

}

getSocketError());

}

throw Exception(parms);

_entry_mut.unlock();

PEG_METHOD_EXIT();

}

PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL Monitor::_dispatch(void *parm)

// Bind the listen socket to the loopback address

{

if (::bind(

HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(parm);

_listenSocket,

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

reinterpret_cast<struct sockaddr*>(&listenAddress),

"Monitor::_dispatch: entering run() for indx = %d, queueId = %d, q = %p",

addressLength) < 0)

dst->_entry_index, dst->_monitor->_entries[dst->_entry_index].queueId, dst);

try

{

dst->run(1);

}

catch (...)

{

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

MessageLoaderParms parms(

"Monitor::_dispatch: exception received");

"Common.Monitor.TICKLE_BIND",

"Received error number $0 while binding the internal socket.",

getSocketError());

throw Exception(parms);

}

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

"Monitor::_dispatch: exited run() for index %d", dst->_entry_index);

PEGASUS_ASSERT(dst->_monitor->_entries[dst->_entry_index]._status.value() == _MonitorEntry::BUSY);

// Listen for a connection from the tickle client

if ((::listen(_listenSocket, 3)) < 0)

// Once the HTTPConnection thread has set the status value to either

// Monitor::DYING or Monitor::IDLE, it has returned control of the connection

// to the Monitor. It is no longer permissible to access the connection

// or the entry in the _entries table.

if (dst->_connectionClosePending)

{

dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::DYING;

MessageLoaderParms parms(

"Common.Monitor.TICKLE_LISTEN",

"Received error number $0 while listening to the internal socket.",

getSocketError());

throw Exception(parms);

}

else

// Verify we have the correct listen socket

SocketLength tmpAddressLength = addressLength;

int sock = ::getsockname(

_listenSocket,

reinterpret_cast<struct sockaddr*>(&listenAddress),

&tmpAddressLength);

if (sock < 0)

{

dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::IDLE;

MessageLoaderParms parms(

}

"Common.Monitor.TICKLE_SOCKNAME",

return 0;

"Received error number $0 while getting the internal socket name.",

getSocketError());

throw Exception(parms);

}

// Set up the client side of the tickle connection.

// Create the client socket

////************************* monitor 2 *****************************////

if ((_clientSocket = Socket::createSocket(addressFamily, SOCK_STREAM, 0)) ==

////************************* monitor 2 *****************************////

PEGASUS_INVALID_SOCKET)

////************************* monitor 2 *****************************////

m2e_rep::m2e_rep(void)

:Base(), state(IDLE)

{

MessageLoaderParms parms(

"Common.Monitor.TICKLE_CLIENT_CREATE",

"Received error number $0 while creating the internal client "

"socket.",

getSocketError());

throw Exception(parms);

}

m2e_rep::m2e_rep(monitor_2_entry_type _type,

// Bind the client socket to the loopback address

pegasus_socket _sock,

if (::bind(

void* _accept,

_clientSocket,

void* _dispatch)

reinterpret_cast<struct sockaddr*>(&clientAddress),

: Base(), type(_type), state(IDLE), psock(_sock),

addressLength) < 0)

accept_parm(_accept), dispatch_parm(_dispatch)

{

MessageLoaderParms parms(

"Common.Monitor.TICKLE_CLIENT_BIND",

"Received error number $0 while binding the internal client "

"socket.",

getSocketError());

throw Exception(parms);

}

m2e_rep::~m2e_rep(void)

// Connect the client socket to the listen socket address

if (::connect(

_clientSocket,

reinterpret_cast<struct sockaddr*>(&listenAddress),

addressLength) < 0)

{

MessageLoaderParms parms(

"Common.Monitor.TICKLE_CLIENT_CONNECT",

"Received error number $0 while connecting the internal client "

"socket.",

getSocketError());

throw Exception(parms);

}

m2e_rep::m2e_rep(const m2e_rep& r)

: Base()

// Set up the server side of the tickle connection.

{

if(this != &r){

type = r.type;

psock = r.psock;

accept_parm = r.accept_parm;

dispatch_parm = r.dispatch_parm;

state = IDLE;

}

tmpAddressLength = addressLength;

}

// Accept the client socket connection.

_serverSocket = ::accept(

_listenSocket,

reinterpret_cast<struct sockaddr*>(&serverAddress),

&tmpAddressLength);

m2e_rep& m2e_rep::operator =(const m2e_rep& r)

if (_serverSocket == PEGASUS_SOCKET_ERROR)

{

if(this != &r) {

MessageLoaderParms parms(

type = r.type;

"Common.Monitor.TICKLE_ACCEPT",

psock = r.psock;

"Received error number $0 while accepting the internal socket "

accept_parm = r.accept_parm;

"connection.",

dispatch_parm = r.dispatch_parm;

getSocketError());

state = IDLE;

throw Exception(parms);

}

return *this;

}

Boolean m2e_rep::operator ==(const m2e_rep& r)

{

// Close the listen socket and make the other sockets non-blocking

if(this == &r)

return true;

return false;

}

Boolean m2e_rep::operator ==(void* r)

Socket::close(_listenSocket);

{

Socket::disableBlocking(_serverSocket);

if((void*)this == r)

Socket::disableBlocking(_clientSocket);

return true;

return false;

}

m2e_rep::operator pegasus_socket() const

{

return psock;

}

#endif

monitor_2_entry::monitor_2_entry(void)

void Tickler::_uninitialize()

{

_rep = new m2e_rep();

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");

}

monitor_2_entry::monitor_2_entry(pegasus_socket& _psock,

try

monitor_2_entry_type _type,

void* _accept_parm, void* _dispatch_parm)

{

_rep = new m2e_rep(_type, _psock, _accept_parm, _dispatch_parm);

Socket::close(_serverSocket);

Socket::close(_clientSocket);

Socket::close(_listenSocket);

}

catch (...)

monitor_2_entry::monitor_2_entry(const monitor_2_entry& e)

{

if(this != &e){

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL2,

Inc(this->_rep = e._rep);

"Failed to close tickle sockets");

}

Socket::uninitializeInterface();

}

monitor_2_entry::~monitor_2_entry(void)

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

// Monitor

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

#define MAX_NUMBER_OF_MONITOR_ENTRIES 32

Monitor::Monitor()

: _stopConnections(0),

_stopConnectionsSem(0),

_solicitSocketCount(0)

{

int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;

_entries.reserveCapacity(numberOfMonitorEntriesToAllocate);

Dec(_rep);

// Create a MonitorEntry for the Tickler and set its state to IDLE so the

}

// Monitor will watch for its events.

_entries.append(MonitorEntry(

_tickler.getReadHandle(),

MonitorEntry::STATUS_IDLE,

MonitorEntry::TYPE_TICKLER));

monitor_2_entry& monitor_2_entry::operator=(const monitor_2_entry& e)

// Start the count at 1 because _entries[0] is the Tickler

for (int i = 1; i < numberOfMonitorEntriesToAllocate; i++)

{

if(this != &e){

_entries.append(MonitorEntry());

Dec(_rep);

Inc(this->_rep = e._rep);

}

return *this;

}

Boolean monitor_2_entry::operator ==(const monitor_2_entry& me) const

Monitor::~Monitor()

{

if(this == &me)

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,

return true;

"returning from monitor destructor");

return false;

}

Boolean monitor_2_entry::operator ==(void* k) const

void Monitor::tickle()

{

if((void *)this == k)

_tickler.notify();

return true;

return false;

}

void Monitor::setState(

monitor_2_entry_type monitor_2_entry::get_type(void) const

Uint32 index,

MonitorEntry::Status status)

{

return _rep->type;

AutoMutex autoEntryMutex(_entriesMutex);

// Set the state to requested state

_entries[index].status = status;

}

void monitor_2_entry::set_type(monitor_2_entry_type t)

void Monitor::run(Uint32 milliseconds)

{

_rep->type = t;

struct timeval tv = {milliseconds/1000, milliseconds%1000*1000};

}

fd_set fdread;

FD_ZERO(&fdread);

AutoMutex autoEntryMutex(_entriesMutex);

monitor_2_entry_state monitor_2_entry::get_state(void) const

ArrayIterator<MonitorEntry> entries(_entries);

{

return (monitor_2_entry_state) _rep->state.value();

}

void monitor_2_entry::set_state(monitor_2_entry_state t)

// Check the stopConnections flag. If set, clear the Acceptor monitor

// entries

if (_stopConnections.get() == 1)

{

_rep->state = t;

for (Uint32 indx = 0; indx < entries.size(); indx++)

}

void* monitor_2_entry::get_accept(void) const

{

return _rep->accept_parm;

if (entries[indx].type == MonitorEntry::TYPE_ACCEPTOR)

}

void monitor_2_entry::set_accept(void* a)

{

_rep->accept_parm = a;

if (entries[indx].status != MonitorEntry::STATUS_EMPTY)

}

void* monitor_2_entry::get_dispatch(void) const

{

return _rep->dispatch_parm;

if (entries[indx].status == MonitorEntry::STATUS_IDLE ||

}

entries[indx].status == MonitorEntry::STATUS_DYING)

void monitor_2_entry::set_dispatch(void* a)

{

_rep->dispatch_parm = a;

// remove the entry

entries[indx].status = MonitorEntry::STATUS_EMPTY;

}

else

pegasus_socket monitor_2_entry::get_sock(void) const

{

return _rep->psock;

// set status to DYING

entries[indx].status = MonitorEntry::STATUS_DYING;

}

void monitor_2_entry::set_sock(pegasus_socket& s)

}

{

_stopConnections = 0;

_rep->psock = s;

_stopConnectionsSem.signal();

}

//static monitor_2* _m2_instance;

for (Uint32 indx = 0; indx < entries.size(); indx++)

AsyncDQueue<HTTPConnection2> monitor_2::_connections(true, 0);

monitor_2::monitor_2(void)

: _session_dispatch(0), _accept_dispatch(0), _listeners(true, 0),

_ready(true, 0), _die(0), _requestCount(0)

{

try {

const MonitorEntry& entry = entries[indx];

bsd_socket_factory _factory;

// set up the listener/acceptor

if ((entry.status == MonitorEntry::STATUS_DYING) &&

pegasus_socket temp = pegasus_socket(&_factory);

(entry.type == MonitorEntry::TYPE_CONNECTION))

{

temp.socket(PF_INET, SOCK_STREAM, 0);

MessageQueue *q = MessageQueue::lookup(entry.queueId);

// initialize the address

PEGASUS_ASSERT(q != 0);

memset(&_tickle_addr, 0, sizeof(_tickle_addr));

HTTPConnection &h = *static_cast<HTTPConnection *>(q);

#ifdef PEGASUS_OS_ZOS

_tickle_addr.sin_addr.s_addr = inet_addr_ebcdic("127.0.0.1");

#else

#ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM

#pragma convert(37)

#endif

_tickle_addr.sin_addr.s_addr = inet_addr("127.0.0.1");

#ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM

#pragma convert(0)

#endif

_tickle_addr.sin_family = PF_INET;

_tickle_addr.sin_port = 0;

PEGASUS_SOCKLEN_SIZE _addr_size = sizeof(_tickle_addr);

temp.bind((struct sockaddr *)&_tickle_addr, sizeof(_tickle_addr));

temp.listen(3);

temp.getsockname((struct sockaddr*)&_tickle_addr, &_addr_size);

// set up the connector

pegasus_socket tickler = pegasus_socket(&_factory);

tickler.socket(PF_INET, SOCK_STREAM, 0);

struct sockaddr_in _addr;

memset(&_addr, 0, sizeof(_addr));

#ifdef PEGASUS_OS_ZOS

_addr.sin_addr.s_addr = inet_addr_ebcdic("127.0.0.1");

#else

_addr.sin_addr.s_addr = inet_addr("127.0.0.1");

#endif

_addr.sin_family = PF_INET;

_addr.sin_port = 0;

tickler.bind((struct sockaddr*)&_addr, sizeof(_addr));

tickler.connect((struct sockaddr*)&_tickle_addr, sizeof(_tickle_addr));

_tickler.set_sock(tickler);

_tickler.set_type(INTERNAL);

_tickler.set_state(BUSY);

struct sockaddr_in peer;

if (h._connectionClosePending == false)

memset(&peer, 0, sizeof(peer));

continue;

PEGASUS_SOCKLEN_SIZE peer_size = sizeof(peer);

pegasus_socket accepted = temp.accept((struct sockaddr*)&peer, &peer_size);

// NOTE: do not attempt to delete while there are pending responses

// coming thru. The last response to come thru after a

// _connectionClosePending will reset _responsePending to false

// and then cause the monitor to rerun this code and clean up.

// (see HTTPConnection.cpp)

if (h._responsePending == true)

{

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

"Monitor::run - Ignoring connection delete request "

"because responses are still pending. "

"connection=0x%p, socket=%d\n",

(void *)&h, h.getSocket()));

continue;

}

h._connectionClosePending = false;

HTTPAcceptor &o = h.getOwningAcceptor();

Message* message= new CloseConnectionMessage(entry.socket);

message->dest = o.getQueueId();

monitor_2_entry* _tickle = new monitor_2_entry(accepted, INTERNAL, 0, 0);

// HTTPAcceptor is responsible for closing the connection.

_tickle->set_state(BUSY);

// The lock is released to allow HTTPAcceptor to call

// unsolicitSocketMessages to free the entry.

// Once HTTPAcceptor completes processing of the close

// connection, the lock is re-requested and processing of

// the for loop continues. This is safe with the current

// implementation of the entries object. Note that the

// loop condition accesses the entries.size() on each

// iteration, so that a change in size while the mutex is

// unlocked will not result in an ArrayIndexOutOfBounds

// exception.

_listeners.insert_first(_tickle);

_entriesMutex.unlock();

o.enqueue(message);

_entriesMutex.lock();

// After enqueue a message and the autoEntryMutex has been

// released and locked again, the array of _entries can be

// changed. The ArrayIterator has be reset with the original

// _entries.

entries.reset(_entries);

}

catch(...){ }

}

monitor_2::~monitor_2(void)

Uint32 _idleEntries = 0;

{

stop();

We will keep track of the maximum socket number and pass this value

to the kernel as a parameter to SELECT. This loop seems like a good

place to calculate the max file descriptor (maximum socket number)

because we have to traverse the entire array.

SocketHandle maxSocketCurrentPass = 0;

for (Uint32 indx = 0; indx < entries.size(); indx++)

{

if (maxSocketCurrentPass < entries[indx].socket)

maxSocketCurrentPass = entries[indx].socket;

try {

if (entries[indx].status == MonitorEntry::STATUS_IDLE)

monitor_2_entry* temp = _listeners.remove_first();

{

while(temp){

_idleEntries++;

delete temp;

FD_SET(entries[indx].socket, &fdread);

temp = _listeners.remove_first();

}

catch(...){ }

Add 1 then assign maxSocket accordingly. We add 1 to account for

descriptors starting at 0.

maxSocketCurrentPass++;

_entriesMutex.unlock();

try

{

// The first argument to select() is ignored on Windows and it is not

HTTPConnection2* temp = _connections.remove_first();

// a socket value. The original code assumed that the number of sockets

while(temp)

// and a socket value have the same type. On Windows they do not.

{

delete temp;

#ifdef PEGASUS_OS_TYPE_WINDOWS

temp = _connections.remove_first();

int events = select(0, &fdread, NULL, NULL, &tv);

}

#else

}

int events = select(maxSocketCurrentPass, &fdread, NULL, NULL, &tv);

catch(...)

#endif

{

int selectErrno = getSocketError();

}

_entriesMutex.lock();

}

struct timeval timeNow;

Time::gettimeofday(&timeNow);

// After enqueue a message and the autoEntryMutex has been released and

// locked again, the array of _entries can be changed. The ArrayIterator

// has be reset with the original _entries

entries.reset(_entries);

void monitor_2::run(void)

if (events == PEGASUS_SOCKET_ERROR)

{

monitor_2_entry* temp;

PEG_TRACE((TRC_HTTP, Tracer::LEVEL1,

while(_die.value() == 0) {

"Monitor::run - select() returned error %d.", selectErrno));

// The EBADF error indicates that one or more or the file

struct timeval tv_idle = { 60, 0 };

// descriptions was not valid. This could indicate that

// the entries structure has been corrupted or that

// we have a synchronization error.

// place all sockets in the select set

PEGASUS_ASSERT(selectErrno != EBADF);

FD_ZERO(&rd_fd_set);

try {

_listeners.lock(pegasus_thread_self());

temp = _listeners.next(0);

while(temp != 0 ){

if(temp->get_state() == CLOSED ) {

monitor_2_entry* closed = temp;

temp = _listeners.next(closed);

_listeners.remove_no_lock(closed);

HTTPConnection2 *cn = monitor_2::remove_connection((Sint32)(closed->get_sock()));

delete cn;

delete closed;

}

if(temp == 0)

else if (events)

break;

{

Sint32 fd = (Sint32) temp->get_sock();

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

if(fd >= 0 )

"Monitor::run select event received events = %d, monitoring %d "

FD_SET(fd , &rd_fd_set);

"idle entries",

temp = _listeners.next(temp);

events, _idleEntries));

}

for (Uint32 indx = 0; indx < entries.size(); indx++)

_listeners.unlock();

{

}

// The Monitor should only look at entries in the table that are

catch(...){

// IDLE (i.e., owned by the Monitor).

return;

if ((entries[indx].status == MonitorEntry::STATUS_IDLE) &&

}

(FD_ISSET(entries[indx].socket, &fdread)))

// important - the dispatch routine has pointers to all the

{

// entries that are readable. These entries can be changed but

MessageQueue* q = MessageQueue::lookup(entries[indx].queueId);

// the pointer must not be tampered with.

PEGASUS_ASSERT(q != 0);

if(_connections.count() )

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

int events = select(FD_SETSIZE, &rd_fd_set, NULL, NULL, NULL);

"Monitor::run indx = %d, queueId = %d, q = %p",

else

indx, entries[indx].queueId, q));

int events = select(FD_SETSIZE, &rd_fd_set, NULL, NULL, &tv_idle);

if(_die.value())

try

{

break;

if (entries[indx].type == MonitorEntry::TYPE_CONNECTION)

}

{

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

"entries[%d].type is TYPE_CONNECTION",

indx));

HTTPConnection *dst =

reinterpret_cast<HTTPConnection *>(q);

dst->_entry_index = indx;

// Update idle start time because we have received some

// data. Any data is good data at this point, and we'll

// keep the connection alive, even if we've exceeded

// the idleConnectionTimeout, which will be checked

// when we call closeConnectionOnTimeout() next.

Time::gettimeofday(&dst->_idleStartTime);

// Check for accept pending (ie. SSL handshake pending)

// or idle connection timeouts for sockets from which

// we received data (avoiding extra queue lookup below).

if (!dst->closeConnectionOnTimeout(&timeNow))

{

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

"Entering HTTPConnection::run() for "

"indx = %d, queueId = %d, q = %p",

indx, entries[indx].queueId, q));

try {

_listeners.lock(pegasus_thread_self());

temp = _listeners.next(0);

while(temp != 0 ){

Sint32 fd = (Sint32) temp->get_sock();

if(fd >= 0 && FD_ISSET(fd, &rd_fd_set)) {

temp->set_state(BUSY);

FD_CLR(fd, &rd_fd_set);

monitor_2_entry* ready = new monitor_2_entry(*temp);

try

{

_ready.insert_first(ready);

dst->run(1);

}

catch(...)

{

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL1,

"Caught exception from "

"HTTPConnection::run()");

}

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,

_requestCount++;

"Exited HTTPConnection::run()");

}

temp = _listeners.next(temp);

}

_listeners.unlock();

}

catch(...){

return;

}

// now handle the sockets that are ready to read

else if (entries[indx].type == MonitorEntry::TYPE_TICKLER)

if(_ready.count())

_dispatch();

else

{

if(_connections.count() == 0 )

_tickler.reset();

_idle_dispatch(_idle_parm);

}

} // while alive

}

else

int monitor_2::solicitSocketMessages(

Sint32 socket,

Uint32 events,

Uint32 queueId,

int type)

{

PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitSocketMessages");

_entry_mut.lock(pegasus_thread_self());

for(int index = 0; index < (int)_entries.size(); index++)

{

try

{

if(_entries[index]._status.value() == monitor_2_entry::EMPTY)

{

_entries[index].socket = socket;

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

//_entries[index].queueId = queueId;

"Non-connection entry, indx = %d, has been "

//_entries[index]._type = type;

"received.",

_entries[index]._status = _MonitorEntry::IDLE;

indx));

_entry_mut.unlock();

int events = 0;

events |= SocketMessage::READ;

Message* msg = new SocketMessage(

entries[indx].socket, events);

entries[indx].status = MonitorEntry::STATUS_BUSY;

_entriesMutex.unlock();

q->enqueue(msg);

_entriesMutex.lock();

return index;

// After enqueue a message and the autoEntryMutex has

// been released and locked again, the array of

// entries can be changed. The ArrayIterator has to be

// reset with the latest _entries.

entries.reset(_entries);

entries[indx].status = MonitorEntry::STATUS_IDLE;

}

catch(...)

{

}

_entry_mut.unlock();

PEG_METHOD_EXIT();

return -1;

}

// else check for accept pending (ie. SSL handshake pending) or

// idle connection timeouts for sockets from which we did not

// receive data.

else if ((entries[indx].status == MonitorEntry::STATUS_IDLE) &&

entries[indx].type == MonitorEntry::TYPE_CONNECTION)

void monitor_2::unsolicitSocketMessages(Sint32 socket)

{

PEG_METHOD_ENTER(TRC_HTTP, "monitor_2::unsolicitSocketMessages");

_entry2_mut.lock(pegasus_thread_self());

for(int index = 0; index < (int)_entries2.size(); index++)

{

if(_entries2[index].socket == socket)

{

_entries2[index]._status = monitor_2_entry::EMPTY;

MessageQueue* q = MessageQueue::lookup(entries[indx].queueId);

_entries2[index].socket = -1;

PEGASUS_ASSERT(q != 0);

break;

HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(q);

dst->_entry_index = indx;

dst->closeConnectionOnTimeout(&timeNow);

}

_entry2_mut.unlock();

PEG_METHOD_EXIT();

}

// else if "events" is zero (ie. select timed out) then we still need

void* monitor_2::set_session_dispatch(void (*dp)(monitor_2_entry*))

// to check if there are any pending SSL handshakes that have timed out.

else

{

void* old = (void *)_session_dispatch;

for (Uint32 indx = 0; indx < entries.size(); indx++)

_session_dispatch = dp;

return old;

}

void* monitor_2::set_accept_dispatch(void (*dp)(monitor_2_entry*))

{

void* old = (void*)_accept_dispatch;

if ((entries[indx].status == MonitorEntry::STATUS_IDLE) &&

_accept_dispatch = dp;

entries[indx].type == MonitorEntry::TYPE_CONNECTION)

return old;

}

void* monitor_2::set_idle_dispatch(void (*dp)(void*))

{

void* old = (void*)_idle_dispatch;

MessageQueue* q = MessageQueue::lookup(entries[indx].queueId);

_idle_dispatch = dp;

PEGASUS_ASSERT(q != 0);

return old;

HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(q);

dst->_entry_index = indx;

dst->closeConnectionOnTimeout(&timeNow);

}

void* monitor_2::set_idle_parm(void* parm)

{

void* old = _idle_parm;

_idle_parm = parm;

return old;

}

//-----------------------------------------------------------------

// Note on deleting the monitor_2_entry nodes:

// Each case: in the switch statement needs to handle the deletion

// of the monitor_2_entry * node differently. A SESSION dispatch

// routine MUST DELETE the entry during its dispatch handling.

// All other dispatch routines MUST NOT delete the entry during the

// dispatch handling, but must allow monitor_2::_dispatch to delete

// the entry.

// The reason is pretty obscure and it is debatable whether or not

// to even bother, but during cimserver shutdown the single monitor_2_entry*

// will leak unless the _session_dispatch routine takes care of deleting it.

// The reason is that a shutdown messages completely stops everything and

// the _session_dispatch routine never returns. So monitor_2::_dispatch is

// never able to do its own cleanup.

// << Mon Oct 13 09:33:33 2003 mdd >>

//-----------------------------------------------------------------

void monitor_2::_dispatch(void)

{

monitor_2_entry* entry;

try

{

entry = _ready.remove_first();

}

catch(...)

{

}

while(entry != 0 ) {

void Monitor::stopListeningForConnections(Boolean wait)

switch(entry->get_type()) {

case INTERNAL:

static char buffer[2];

entry->get_sock().disableBlocking();

entry->get_sock().read(&buffer, 2);

entry->get_sock().enableBlocking();

delete entry;

break;

case LISTEN:

{

static struct sockaddr peer;

static PEGASUS_SOCKLEN_SIZE peer_size = sizeof(peer);

entry->get_sock().disableBlocking();

pegasus_socket connected = entry->get_sock().accept(&peer, &peer_size);

#ifdef PEGASUS_OS_TYPE_WINDOWS

if((Sint32)connected == SOCKET_ERROR)

#else

if((Sint32)connected == -1 )

#endif

{

delete entry;

PEG_METHOD_ENTER(TRC_HTTP, "Monitor::stopListeningForConnections()");

break;

// set boolean then tickle the server to recognize _stopConnections

}

_stopConnections = 1;

tickle();

entry->get_sock().enableBlocking();

monitor_2_entry *temp = add_entry(connected, SESSION, entry->get_accept(), entry->get_dispatch());

if(temp && _accept_dispatch != 0)

_accept_dispatch(temp);

delete entry;

}

if (wait)

break;

case SESSION:

if(_session_dispatch != 0 )

{

// NOTE: _session_dispatch will delete entry - do not do it here

// Wait for the monitor to notice _stopConnections. Otherwise the

_session_dispatch(entry);

// caller of this function may unbind the ports while the monitor

// is still accepting connections on them.

_stopConnectionsSem.wait();

}

else {

PEG_METHOD_EXIT();

static char buffer[4096];

int bytes = entry->get_sock().read(&buffer, 4096);

delete entry;

}

break;

case UNTYPED:

default:

delete entry;

break;

}

_requestCount--;

if(_ready.count() == 0 )

int Monitor::solicitSocketMessages(

break;

SocketHandle socket,

Uint32 events,

try

Uint32 queueId,

{

Uint32 type)

entry = _ready.remove_first();

}

catch(...)

{

}

PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitSocketMessages");

AutoMutex autoMut(_entriesMutex);

}

// Check to see if we need to dynamically grow the _entries array

}

// We always want the _entries array to be 2 bigger than the

// current connections requested

_solicitSocketCount++; // bump the count

void monitor_2::stop(void)

for (Uint32 i = _entries.size(); i < _solicitSocketCount + 1; i++)

{

_die = 1;

_entries.append(MonitorEntry());

tickle();

// shut down the listener list, free the list nodes

_tickler.get_sock().close();

_listeners.shutdown_queue();

}

void monitor_2::tickle(void)

for (Uint32 index = 1; index < _entries.size(); index++)

{

static char _buffer[] =

try

{

'0','0'

if (_entries[index].status == MonitorEntry::STATUS_EMPTY)

};

_tickler.get_sock().disableBlocking();

_tickler.get_sock().write(&_buffer, 2);

_tickler.get_sock().enableBlocking();

}

monitor_2_entry* monitor_2::add_entry(pegasus_socket& ps,

monitor_2_entry_type type,

void* accept_parm,

void* dispatch_parm)

{

monitor_2_entry* m2e = new monitor_2_entry(ps, type, accept_parm, dispatch_parm);

_entries[index].socket = socket;

_entries[index].queueId = queueId;

_entries[index].type = type;

_entries[index].status = MonitorEntry::STATUS_IDLE;

try{

return (int)index;

_listeners.insert_first(m2e);

}

catch(...){

delete m2e;

return 0;

}

tickle();

return m2e;

}

catch (...)

Boolean monitor_2::remove_entry(Sint32 s)

{

monitor_2_entry* temp;

try {

_listeners.try_lock(pegasus_thread_self());

temp = _listeners.next(0);

while(temp != 0){

if(s == (Sint32)temp->_rep->psock ){

temp = _listeners.remove_no_lock(temp);

delete temp;

_listeners.unlock();

return true;

}

temp = _listeners.next(temp);

}

_listeners.unlock();

}

catch(...){

// decrease the count, if we are here we didn't do anything meaningful

}

_solicitSocketCount--;

return false;

PEG_METHOD_EXIT();

return -1;

}

Uint32 monitor_2::getOutstandingRequestCount(void)

void Monitor::unsolicitSocketMessages(SocketHandle socket)

{

return _requestCount.value();

}

HTTPConnection2* monitor_2::remove_connection(Sint32 sock)

{

PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");

AutoMutex autoMut(_entriesMutex);

HTTPConnection2* temp;

try

Start at index = 1 because _entries[0] is the tickle entry which

never needs to be reset to EMPTY;

for (Uint32 index = 1; index < _entries.size(); index++)

{

monitor_2::_connections.lock(pegasus_thread_self());

if (_entries[index].socket == socket)

temp = monitor_2::_connections.next(0);

while(temp != 0 )

{

if(sock == temp->getSocket())

{

temp = monitor_2::_connections.remove_no_lock(temp);

monitor_2::_connections.unlock();

return temp;

}

temp = monitor_2::_connections.next(temp);

}

monitor_2::_connections.unlock();

}

catch(...)

{

_entries[index].reset();

_solicitSocketCount--;

break;

}

return 0;

}

Boolean monitor_2::insert_connection(HTTPConnection2* connection)

{

Dynamic Contraction:

try

To remove excess entries we will start from the end of the _entries

array and remove all entries with EMPTY status until we find the

first NON EMPTY. This prevents the positions, of the NON EMPTY

entries, from being changed.

for (Uint32 index = _entries.size() - 1;

(_entries[index].status == MonitorEntry::STATUS_EMPTY) &&

(index >= MAX_NUMBER_OF_MONITOR_ENTRIES);

index--)

{

monitor_2::_connections.insert_first(connection);

_entries.remove(index);

}

catch(...)

{

return false;

}

return true;

}

PEG_METHOD_EXIT();

}

PEGASUS_NAMESPACE_END

Legend:

Removed from v.1.70
changed lines
	Added in v.1.135

No CVS admin address has been configured