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.13 and 1.130

version 1.13, 2002/05/28 20:05:35

version 1.130, 2008/02/27 20:21:17

Line 1

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

//%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 Brasher (mbrasher@bmc.com)

// Modified By:

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

#include "Network.h"

#include <Pegasus/Common/Config.h>

#include <cstring>

#include "Monitor.h"

Line 33

Line 39

#include "Socket.h"

#include <Pegasus/Common/Tracer.h>

#include <Pegasus/Common/HTTPConnection.h>

#include <Pegasus/Common/MessageQueueService.h>

#ifdef PEGASUS_OS_TYPE_WINDOWS

#include <Pegasus/Common/Exception.h>

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

#include "ArrayIterator.h"

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

#include "HostAddress.h"

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

<windows.h>). Find the 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>

# include <unistd.h>

#endif

PEGASUS_USING_STD;

Line 62

Line 50

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

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

_initialize();

fd_set active_rd_fd_set;

}

fd_set active_wr_fd_set;

catch (...)

fd_set active_ex_fd_set;

{

};

_uninitialize();

throw;

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

}

// Monitor

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

Monitor::Monitor()

Tickler::~Tickler()

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

{

Socket::initializeInterface();

_uninitialize();

_rep = new MonitorRep;

FD_ZERO(&_rep->rd_fd_set);

FD_ZERO(&_rep->wr_fd_set);

FD_ZERO(&_rep->ex_fd_set);

FD_ZERO(&_rep->active_rd_fd_set);

FD_ZERO(&_rep->active_wr_fd_set);

FD_ZERO(&_rep->active_ex_fd_set);

}

Monitor::~Monitor()

#if defined(PEGASUS_OS_TYPE_UNIX)

// Use an anonymous pipe for the tickle connection.

void Tickler::_initialize()

{

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

int fds[2];

"deregistering with module controller");

if(_module_handle != NULL)

if (pipe(fds) == -1)

{

_controller->deregister_module(PEGASUS_MODULENAME_MONITOR);

MessageLoaderParms parms(

_controller = 0;

"Common.Monitor.TICKLE_CREATE",

delete _module_handle;

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

getSocketError());

throw Exception(parms);

}

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

delete _rep;

_serverSocket = fds[0];

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

_clientSocket = fds[1];

Socket::uninitializeInterface();

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

"returning from monitor destructor");

}

#else

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

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

//<<< Tue May 14 20:38:26 2002 mdd >>>

void Tickler::_initialize()

// register with module controller

// when it is time to enqueue the message,

// use an async_thread_exec call to

// isolate the entire if(events) { enqueue -> fd_clear } block

// let the thread pool grow and shrink according to load.

Boolean Monitor::run(Uint32 milliseconds)

{

// register the monitor as a module to gain access to the cimserver's thread pool

// <<< Wed May 15 09:52:16 2002 mdd >>>

while( _module_handle == NULL)

{

try

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

// based on whether IPv6 is enabled.

Socket::initializeInterface();

# ifdef PEGASUS_ENABLE_IPV6

struct sockaddr_storage listenAddress;

struct sockaddr_storage clientAddress;

struct sockaddr_storage serverAddress;

# else

struct sockaddr_in listenAddress;

struct sockaddr_in clientAddress;

struct sockaddr_in serverAddress;

# endif

int addressFamily;

SocketLength addressLength;

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

# ifdef PEGASUS_ENABLE_IPV6

if (System::isIPv6StackActive())

{

_controller = &(ModuleController::register_module(PEGASUS_QUEUENAME_CONTROLSERVICE,

// Use the IPv6 loopback address for the listen sockets

PEGASUS_MODULENAME_MONITOR,

HostAddress::convertTextToBinary(

(void *)this,

HostAddress::AT_IPV6,

"::1",

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

listenAddress.ss_family = AF_INET6;

&_module_handle));

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

addressFamily = AF_INET6;

addressLength = sizeof(struct sockaddr_in6);

}

catch(IncompatibleTypes &)

else

# endif

{

ModuleController* controlService =

// Use the IPv4 loopback address for the listen sockets

new ModuleController(PEGASUS_QUEUENAME_CONTROLSERVICE);

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;

addressLength = sizeof(struct sockaddr_in);

}

catch( AlreadyExists & )

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

clientAddress = listenAddress;

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

// Create the listen socket

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

PEGASUS_INVALID_SOCKET)

{

break;

MessageLoaderParms parms(

"Common.Monitor.TICKLE_CREATE",

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

getSocketError());

throw Exception(parms);

}

// Bind the listen socket to the loopback address

if (::bind(

_listenSocket,

reinterpret_cast<struct sockaddr*>(&listenAddress),

addressLength) < 0)

{

MessageLoaderParms parms(

"Common.Monitor.TICKLE_BIND",

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

getSocketError());

throw Exception(parms);

}

#ifdef PEGASUS_OS_TYPE_WINDOWS

// Listen for a connection from the tickle client

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

{

MessageLoaderParms parms(

"Common.Monitor.TICKLE_LISTEN",

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

getSocketError());

throw Exception(parms);

}

// Windows select() has a strange little bug. It returns immediately if

// Verify we have the correct listen socket

// there are no descriptors in the set even if the timeout is non-zero.

SocketLength tmpAddressLength = addressLength;

// To work around this, we call Sleep() for now:

int sock = ::getsockname(

_listenSocket,

reinterpret_cast<struct sockaddr*>(&listenAddress),

&tmpAddressLength);

if (sock < 0)

{

MessageLoaderParms parms(

"Common.Monitor.TICKLE_SOCKNAME",

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

getSocketError());

throw Exception(parms);

}

if (_entries.size() == 0)

Sleep(milliseconds);

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

#endif

// Create the client socket

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

PEGASUS_INVALID_SOCKET)

{

MessageLoaderParms parms(

"Common.Monitor.TICKLE_CLIENT_CREATE",

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

"socket.",

getSocketError());

throw Exception(parms);

}

// Check for events on the selected file descriptors. Only do this if

// Bind the client socket to the loopback address

// there were no undispatched events from last time.

if (::bind(

_clientSocket,

reinterpret_cast<struct sockaddr*>(&clientAddress),

addressLength) < 0)

{

MessageLoaderParms parms(

"Common.Monitor.TICKLE_CLIENT_BIND",

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

"socket.",

getSocketError());

throw Exception(parms);

}

int count = 0;

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

}

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

memcpy(&_rep->active_rd_fd_set, &_rep->rd_fd_set, sizeof(fd_set));

tmpAddressLength = addressLength;

memcpy(&_rep->active_wr_fd_set, &_rep->wr_fd_set, sizeof(fd_set));

memcpy(&_rep->active_ex_fd_set, &_rep->ex_fd_set, sizeof(fd_set));

const Uint32 SECONDS = milliseconds / 1000;

// Accept the client socket connection.

const Uint32 MICROSECONDS = (milliseconds % 1000) * 1000;

_serverSocket = ::accept(

struct timeval tv = { SECONDS, MICROSECONDS };

_listenSocket,

reinterpret_cast<struct sockaddr*>(&serverAddress),

&tmpAddressLength);

count = select(

if (_serverSocket == PEGASUS_SOCKET_ERROR)

FD_SETSIZE,

&_rep->active_rd_fd_set,

&_rep->active_wr_fd_set,

&_rep->active_ex_fd_set,

&tv);

if (count == 0)

{

pegasus_sleep(milliseconds);

MessageLoaderParms parms(

"Common.Monitor.TICKLE_ACCEPT",

return false;

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

"connection.",

getSocketError());

throw Exception(parms);

}

#ifdef PEGASUS_OS_TYPE_WINDOWS

else if (count == SOCKET_ERROR)

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

#else

else if (count == -1)

#endif

Socket::close(_listenSocket);

{

Socket::disableBlocking(_serverSocket);

pegasus_sleep(milliseconds);

Socket::disableBlocking(_clientSocket);

return false;

}

Boolean handled_events = false;

#endif

for (Uint32 i = 0, n = _entries.size(); i < _entries.size(); i++)

{

Sint32 socket = _entries[i].socket;

Uint32 events = 0;

if(_entries[i].dying.value() > 0 )

void Tickler::_uninitialize()

{

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

{

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

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

try

if(q && static_cast<HTTPConnection *>(q)->is_dying() &&

(0 == static_cast<HTTPConnection *>(q)->refcount.value()))

{

static_cast<HTTPConnection *>(q)->lock_connection();

Socket::close(_serverSocket);

static_cast<HTTPConnection *>(q)->unlock_connection();

Socket::close(_clientSocket);

Socket::close(_listenSocket);

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

Message* message= new CloseConnectionMessage(static_cast<HTTPConnection *>(q)->getSocket());

message->dest = o.getQueueId();

o.enqueue(message);

i--;

n = _entries.size();

}

catch (...)

{

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,

"Failed to close tickle sockets");

}

Socket::uninitializeInterface();

}

if (FD_ISSET(socket, &_rep->active_rd_fd_set))

events |= SocketMessage::READ;

if (FD_ISSET(socket, &_rep->active_wr_fd_set))

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

events |= SocketMessage::WRITE;

// Monitor

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

if (FD_ISSET(socket, &_rep->active_ex_fd_set))

#define MAX_NUMBER_OF_MONITOR_ENTRIES 32

events |= SocketMessage::EXCEPTION;

Monitor::Monitor()

: _stopConnections(0),

_stopConnectionsSem(0),

_solicitSocketCount(0)

{

int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;

_entries.reserveCapacity(numberOfMonitorEntriesToAllocate);

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

// Monitor will watch for its events.

_entries.append(MonitorEntry(

_tickler.getServerSocket(),

MonitorEntry::STATUS_IDLE,

MonitorEntry::TYPE_INTERNAL));

if (events)

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

{

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

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

"Monitor::run - Socket Event Detected events = %d", events);

if (events & SocketMessage::WRITE)

{

FD_CLR(socket, &_rep->active_wr_fd_set);

_entries.append(MonitorEntry());

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

"Monitor::run FD_CLR WRITE");

}

if (events & SocketMessage::EXCEPTION)

}

Monitor::~Monitor()

{

FD_CLR(socket, &_rep->active_ex_fd_set);

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,

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

"returning from monitor destructor");

"Monitor::run FD_CLR EXECEPTION");

}

if (events & SocketMessage::READ)

void Monitor::tickle()

{

FD_CLR(socket, &_rep->active_rd_fd_set);

Socket::write(_tickler.getClientSocket(), "\0", 1);

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

"Monitor::run FD_CLR READ");

}

MessageQueue* queue = MessageQueue::lookup(_entries[i].queueId);

if( ! queue )

void Monitor::setState(

Uint32 index,

MonitorEntry::Status status)

{

unsolicitSocketMessages(socket);

AutoMutex autoEntryMutex(_entriesMutex);

break;

// Set the state to requested state

_entries[index].status = status;

}

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

void Monitor::run(Uint32 milliseconds)

{

if( static_cast<HTTPConnection *>(queue)->refcount.value() == 0 )

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

fd_set fdread;

FD_ZERO(&fdread);

AutoMutex autoEntryMutex(_entriesMutex);

ArrayIterator<MonitorEntry> entries(_entries);

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

// entries

if (_stopConnections.get() == 1)

{

static_cast<HTTPConnection *>(queue)->refcount++;

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

if( false == static_cast<HTTPConnection *>(queue)->is_dying())

{

_controller->async_thread_exec(*_module_handle, _dispatch, (void *)queue);

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

else

{

static_cast<HTTPConnection *>(queue)->refcount--;

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

}

{

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

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

{

// remove the entry

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

}

else

{

Message* message = new SocketMessage(socket, events);

// set status to DYING

queue->enqueue(message);

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

}

count--;

}

handled_events = true;

}

return(handled_events);

_stopConnections = 0;

_stopConnectionsSem.signal();

}

Boolean Monitor::solicitSocketMessages(

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

Sint32 socket,

{

Uint32 events,

const MonitorEntry& entry = entries[indx];

Uint32 queueId,

int type)

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

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

{

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

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

PEGASUS_ASSERT(q != 0);

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

// See whether a handler is already registered for this one:

if (h._connectionClosePending == false)

Uint32 pos = _findEntry(socket);

continue;

if (pos != PEGASUS_NOT_FOUND)

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

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

return false;

"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;

MessageQueue &o = h.get_owner();

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

message->dest = o.getQueueId();

// Set the events:

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

if (events & SocketMessage::READ)

_entriesMutex.unlock();

FD_SET(socket, &_rep->rd_fd_set);

o.enqueue(message);

_entriesMutex.lock();

if (events & SocketMessage::WRITE)

// After enqueue a message and the autoEntryMutex has been

FD_SET(socket, &_rep->wr_fd_set);

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

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

// _entries.

entries.reset(_entries);

}

if (events & SocketMessage::EXCEPTION)

Uint32 _idleEntries = 0;

FD_SET(socket, &_rep->ex_fd_set);

// Add the entry to the list:

_MonitorEntry entry(socket, queueId, type);

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

_entries.append(entry);

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;

// Success!

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

{

_idleEntries++;

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

}

PEG_METHOD_EXIT();

return true;

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

descriptors starting at 0.

maxSocketCurrentPass++;

_entriesMutex.unlock();

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

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

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

#ifdef PEGASUS_OS_TYPE_WINDOWS

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

#else

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

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

if (events == PEGASUS_SOCKET_ERROR)

{

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

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

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

// descriptions was not valid. This could indicate that

// the entries structure has been corrupted or that

// we have a synchronization error.

PEGASUS_ASSERT(selectErrno != EBADF);

}

else if (events)

{

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

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

"idle entries",

events, _idleEntries));

for (Uint32 indx = 0; indx < 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 == MonitorEntry::STATUS_IDLE) &&

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

{

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

PEGASUS_ASSERT(q != 0);

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

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

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

try

{

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

{

dst->run(1);

}

catch (...)

{

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL2,

"Caught exception from "

"HTTPConnection::run()");

}

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,

"Exited HTTPConnection::run()");

}

Boolean Monitor::unsolicitSocketMessages(Sint32 socket)

}

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

{

char buffer;

Sint32 ignored =

Socket::read(entries[indx].socket, &buffer, 1);

}

else

{

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

PEG_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::STATUS_BUSY;

_entriesMutex.unlock();

q->enqueue(msg);

_entriesMutex.lock();

// Look for the given entry and remove it:

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

{

}

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

for (Uint32 i = 0, n = _entries.size(); i < n; i++)

{

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

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

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

dst->_entry_index = indx;

dst->closeConnectionOnTimeout(&timeNow);

}

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

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

else

{

Sint32 socket = _entries[i].socket;

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

FD_CLR(socket, &_rep->rd_fd_set);

{

FD_CLR(socket, &_rep->wr_fd_set);

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

FD_CLR(socket, &_rep->ex_fd_set);

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

_entries.remove(i);

{

// ATTN-RK-P3-20020521: Need "Socket::close(socket);" here?

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

PEG_METHOD_EXIT();

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

return true;

dst->_entry_index = indx;

dst->closeConnectionOnTimeout(&timeNow);

}

PEG_METHOD_EXIT();

return false;

}

Uint32 Monitor::_findEntry(Sint32 socket)

void Monitor::stopListeningForConnections(Boolean wait)

{

for (Uint32 i = 0, n = _entries.size(); i < n; i++)

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

// set boolean then tickle the server to recognize _stopConnections

_stopConnections = 1;

tickle();

if (wait)

{

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

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

return i;

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

// is still accepting connections on them.

_stopConnectionsSem.wait();

}

return PEG_NOT_FOUND;

PEG_METHOD_EXIT();

}

PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL Monitor::_dispatch(void *parm)

int Monitor::solicitSocketMessages(

SocketHandle socket,

Uint32 events,

Uint32 queueId,

Uint32 type)

{

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

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

if( true == dst->is_dying())

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

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

{

dst->refcount--;

_entries.append(MonitorEntry());

return 0;

}

if( false == dst->is_dying())

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

{

dst->run(1);

try

{

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

{

_entries[index].socket = socket;

_entries[index].queueId = queueId;

_entries[index].type = type;

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

return (int)index;

}

catch (...)

{

}

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

_solicitSocketCount--;

PEG_METHOD_EXIT();

return -1;

}

dst->refcount--;

return 0;

void Monitor::unsolicitSocketMessages(SocketHandle socket)

{

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

AutoMutex autoMut(_entriesMutex);

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

{

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

{

_entries[index].reset();

_solicitSocketCount--;

break;

}

Dynamic Contraction:

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

{

_entries.remove(index);

}

PEG_METHOD_EXIT();

}

PEGASUS_NAMESPACE_END

Legend:

Removed from v.1.13
changed lines
	Added in v.1.130

No CVS admin address has been configured